| 84 | |
| 85 | |
| 86 | class WordPressMigrator: |
| 87 | def __init__(self, site_url: str, output_dir: str, delay: float = 0.3, |
| 88 | dry_run: bool = False, url_filter: Optional[str] = None): |
| 89 | self.site_url = site_url.rstrip("/") |
| 90 | self.output_dir = Path(output_dir) |
| 91 | self.delay = delay |
| 92 | self.dry_run = dry_run |
| 93 | self.url_filter = url_filter |
| 94 | self.session = requests.Session() |
| 95 | self.session.headers.update({ |
| 96 | "User-Agent": "docmcp-knowledge-migrator/1.0" |
| 97 | }) |
| 98 | self.h2t = html2text.HTML2Text() |
| 99 | self.h2t.ignore_links = False |
| 100 | self.h2t.ignore_images = False |
| 101 | self.h2t.body_width = 0 |
| 102 | self.h2t.protect_links = True |
| 103 | self.h2t.unicode_snob = True |
| 104 | self.stats: Dict[str, int] = { |
| 105 | "total": 0, "success": 0, "skipped": 0, "error": 0 |
| 106 | } |
| 107 | |
| 108 | def api_get(self, endpoint: str, params: Optional[dict] = None) -> Optional[object]: |
| 109 | url = f"{self.site_url}/wp-json/wp/v2/{endpoint}" |
| 110 | try: |
| 111 | resp = self.session.get(url, params=params, timeout=30) |
| 112 | if resp.status_code == 400: |
| 113 | return None |
| 114 | resp.raise_for_status() |
| 115 | return resp.json() |
| 116 | except requests.RequestException as e: |
| 117 | print(f" ERROR fetching {url}: {e}") |
| 118 | return None |
| 119 | |
| 120 | def get_all_docs(self) -> List[dict]: |
| 121 | """Fetch all published docs with pagination.""" |
| 122 | all_docs = [] |
| 123 | page = 1 |
| 124 | params: dict = { |
| 125 | "per_page": 100, |
| 126 | "status": "publish", |
| 127 | "_fields": "id,title,slug,link,content,parent,modified", |
| 128 | } |
| 129 | while True: |
| 130 | params["page"] = page |
| 131 | docs = self.api_get("docs", params) |
| 132 | if not docs or not isinstance(docs, list): |
| 133 | break |
| 134 | all_docs.extend(docs) |
| 135 | print(f" Fetched page {page}: {len(docs)} docs (total: {len(all_docs)})") |
| 136 | if len(docs) < 100: |
| 137 | break |
| 138 | page += 1 |
| 139 | time.sleep(self.delay) |
| 140 | return all_docs |
| 141 | |
| 142 | def detect_output_path(self, link: str) -> Tuple[str, str]: |
| 143 | """ |