eb46e739e7
Playwright-based crawler that renders pages from api.intra-mart.jp and document.intra-mart.jp, extracts main content, and converts to Markdown via Pandoc. SQLite-backed queue drives a resumable sequential pipeline across crawler, extractor, converter, and storage modules. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
557 lines
11 KiB
Python
557 lines
11 KiB
Python
import json
|
|
from urllib.parse import urljoin
|
|
|
|
from bs4 import BeautifulSoup
|
|
|
|
from utils.logger import logger
|
|
|
|
|
|
def _ensure_soup(html_or_soup):
|
|
if hasattr(html_or_soup, "select_one") and hasattr(
|
|
html_or_soup, "find_all"
|
|
):
|
|
return html_or_soup
|
|
return BeautifulSoup(html_or_soup, "lxml")
|
|
|
|
|
|
|
|
class NavigationParser:
|
|
"""
|
|
Parse documentation navigation/sidebar
|
|
to preserve hierarchy structure.
|
|
|
|
Supports:
|
|
- sidebar menus
|
|
- nested navigation trees
|
|
- table of contents
|
|
- navigation JSON
|
|
"""
|
|
|
|
def __init__(self, url_manager=None):
|
|
self.discovered_urls = set()
|
|
self.url_manager = url_manager
|
|
|
|
def parse_navigation(
|
|
self,
|
|
base_url: str,
|
|
html
|
|
):
|
|
"""
|
|
Main entry point. Accepts an HTML string or a parsed
|
|
BeautifulSoup/Tag.
|
|
|
|
Returns:
|
|
-------
|
|
{
|
|
"urls": [...],
|
|
"tree": [...]
|
|
}
|
|
"""
|
|
|
|
soup = _ensure_soup(html)
|
|
|
|
urls = set()
|
|
|
|
tree = []
|
|
|
|
# Parse sidebar menus
|
|
sidebar_tree = self.parse_sidebar_navigation(
|
|
soup,
|
|
base_url
|
|
)
|
|
|
|
tree.extend(sidebar_tree)
|
|
|
|
urls.update(
|
|
self.extract_urls_from_tree(sidebar_tree)
|
|
)
|
|
|
|
sphinx_tree = self.parse_sphinx_toctree(
|
|
soup,
|
|
base_url
|
|
)
|
|
|
|
tree.extend(sphinx_tree)
|
|
|
|
urls.update(
|
|
self.extract_urls_from_tree(sphinx_tree)
|
|
)
|
|
|
|
# Parse TOC
|
|
toc_tree = self.parse_toc_navigation(
|
|
soup,
|
|
base_url
|
|
)
|
|
|
|
tree.extend(toc_tree)
|
|
|
|
urls.update(
|
|
self.extract_urls_from_tree(toc_tree)
|
|
)
|
|
|
|
|
|
breadcrumb_tree = self.parse_breadcrumbs(
|
|
soup,
|
|
base_url
|
|
)
|
|
|
|
tree.extend(breadcrumb_tree)
|
|
|
|
urls.update(
|
|
self.extract_urls_from_tree(breadcrumb_tree)
|
|
)
|
|
|
|
# Parse generic nav blocks
|
|
nav_tree = self.parse_generic_navigation(
|
|
soup,
|
|
base_url
|
|
)
|
|
|
|
tree.extend(nav_tree)
|
|
|
|
urls.update(
|
|
self.extract_urls_from_tree(nav_tree)
|
|
)
|
|
|
|
logger.info(
|
|
f"Navigation parser found {len(urls)} URLs"
|
|
)
|
|
|
|
tree = self.dedupe_tree(tree)
|
|
|
|
return {
|
|
"urls": list(urls),
|
|
"tree": tree
|
|
}
|
|
|
|
def parse_sidebar_navigation(
|
|
self,
|
|
soup,
|
|
base_url
|
|
):
|
|
"""
|
|
Parse sidebar-based docs navigation
|
|
"""
|
|
|
|
selectors = [
|
|
"nav",
|
|
".sidebar",
|
|
".navigation",
|
|
".menu",
|
|
".toc",
|
|
".md-nav",
|
|
".wy-menu",
|
|
".theme-doc-sidebar-container",
|
|
".theme-doc-sidebar-menu",
|
|
".sphinxsidebar",
|
|
".sphinxsidebarwrapper",
|
|
".toctree-wrapper",
|
|
".globaltoc",
|
|
".localtoc",
|
|
".relations",
|
|
".related",
|
|
".body .toctree-wrapper",
|
|
".document .toctree-wrapper",
|
|
".contents.topic"
|
|
]
|
|
|
|
results = []
|
|
|
|
for selector in selectors:
|
|
|
|
nodes = soup.select(selector)
|
|
|
|
for node in nodes:
|
|
|
|
tree = self.parse_list_structure(
|
|
node,
|
|
base_url
|
|
)
|
|
|
|
if tree:
|
|
results.extend(tree)
|
|
|
|
return results
|
|
|
|
def parse_toc_navigation(
|
|
self,
|
|
soup,
|
|
base_url
|
|
):
|
|
"""
|
|
Parse table-of-contents sections
|
|
"""
|
|
|
|
selectors = [
|
|
".toc",
|
|
".table-of-contents",
|
|
"#table-of-contents",
|
|
".contents",
|
|
".markdown-toc"
|
|
]
|
|
|
|
results = []
|
|
|
|
for selector in selectors:
|
|
|
|
nodes = soup.select(selector)
|
|
|
|
for node in nodes:
|
|
|
|
tree = self.parse_list_structure(
|
|
node,
|
|
base_url
|
|
)
|
|
|
|
if tree:
|
|
results.extend(tree)
|
|
|
|
return results
|
|
|
|
def parse_generic_navigation(
|
|
self,
|
|
soup,
|
|
base_url
|
|
):
|
|
"""
|
|
Fallback parser:
|
|
extract all nav-like links
|
|
"""
|
|
|
|
results = []
|
|
|
|
for a in soup.find_all("a", href=True):
|
|
|
|
href = a["href"].strip()
|
|
|
|
title = a.get_text(" ", strip=True)
|
|
|
|
if not title:
|
|
continue
|
|
|
|
if not href:
|
|
continue
|
|
|
|
if href.startswith("#"):
|
|
continue
|
|
|
|
if href.startswith("javascript:"):
|
|
continue
|
|
|
|
if href.startswith("mailto:"):
|
|
continue
|
|
|
|
full_url = self.resolve_url(base_url, href)
|
|
if not full_url:
|
|
continue
|
|
|
|
item = {
|
|
"title": title,
|
|
"url": full_url,
|
|
"children": []
|
|
}
|
|
|
|
results.append(item)
|
|
|
|
return results
|
|
|
|
def parse_list_structure(
|
|
self,
|
|
node,
|
|
base_url
|
|
):
|
|
"""
|
|
Parse nested UL/LI navigation tree
|
|
"""
|
|
|
|
results = []
|
|
|
|
root_lists = node.find_all(
|
|
["ul", "ol"],
|
|
recursive=False
|
|
)
|
|
|
|
if not root_lists:
|
|
root_lists = [node]
|
|
|
|
for ul in root_lists:
|
|
|
|
for li in ul.find_all("li", recursive=False):
|
|
|
|
item = self.parse_list_item(
|
|
li,
|
|
base_url
|
|
)
|
|
|
|
if item:
|
|
results.append(item)
|
|
|
|
return results
|
|
|
|
def parse_list_item(
|
|
self,
|
|
li,
|
|
base_url
|
|
):
|
|
"""
|
|
Parse single navigation item
|
|
"""
|
|
|
|
a = li.find("a", href=True)
|
|
|
|
if not a:
|
|
return None
|
|
|
|
href = a["href"].strip()
|
|
|
|
if not href:
|
|
return None
|
|
|
|
full_url = self.resolve_url(base_url, href)
|
|
|
|
if not full_url:
|
|
return None
|
|
|
|
title = a.get_text(" ", strip=True)
|
|
|
|
if not title:
|
|
title = full_url.rsplit("/", 1)[-1]
|
|
|
|
item = {
|
|
"title": title,
|
|
"url": full_url,
|
|
"children": []
|
|
}
|
|
|
|
nested_lists = li.find_all(
|
|
["ul", "ol"],
|
|
recursive=False
|
|
)
|
|
|
|
for nested in nested_lists:
|
|
|
|
for child_li in nested.find_all(
|
|
"li",
|
|
recursive=False
|
|
):
|
|
|
|
child_item = self.parse_list_item(
|
|
child_li,
|
|
base_url
|
|
)
|
|
|
|
if child_item:
|
|
item["children"].append(
|
|
child_item
|
|
)
|
|
|
|
return item
|
|
|
|
def extract_urls_from_tree(
|
|
self,
|
|
tree
|
|
):
|
|
"""
|
|
Flatten navigation tree into URL list
|
|
"""
|
|
|
|
urls = set()
|
|
|
|
for item in tree:
|
|
|
|
url = item.get("url")
|
|
|
|
if url:
|
|
urls.add(url)
|
|
|
|
children = item.get("children", [])
|
|
|
|
child_urls = self.extract_urls_from_tree(
|
|
children
|
|
)
|
|
|
|
urls.update(child_urls)
|
|
|
|
return urls
|
|
|
|
def export_tree_json(
|
|
self,
|
|
tree,
|
|
output_path
|
|
):
|
|
"""
|
|
Export navigation hierarchy to JSON
|
|
"""
|
|
|
|
with open(
|
|
output_path,
|
|
"w",
|
|
encoding="utf-8"
|
|
) as f:
|
|
|
|
json.dump(
|
|
tree,
|
|
f,
|
|
ensure_ascii=False,
|
|
indent=2
|
|
)
|
|
|
|
logger.info(
|
|
f"Navigation tree saved: {output_path}"
|
|
)
|
|
|
|
def generate_summary_markdown(
|
|
self,
|
|
tree
|
|
):
|
|
"""
|
|
Generate SUMMARY.md
|
|
for mdBook / docs navigation
|
|
"""
|
|
|
|
lines = ["# Summary", ""]
|
|
|
|
self.build_summary_lines(
|
|
tree,
|
|
lines,
|
|
level=0
|
|
)
|
|
|
|
return "\n".join(lines)
|
|
|
|
def build_summary_lines(
|
|
self,
|
|
tree,
|
|
lines,
|
|
level=0
|
|
):
|
|
"""
|
|
Recursive markdown tree builder
|
|
"""
|
|
|
|
indent = " " * level
|
|
|
|
for item in tree:
|
|
|
|
title = item.get("title", "Untitled")
|
|
|
|
url = item.get("url", "#")
|
|
|
|
lines.append(
|
|
f"{indent}- [{title}]({url})"
|
|
)
|
|
|
|
children = item.get("children", [])
|
|
|
|
if children:
|
|
|
|
self.build_summary_lines(
|
|
children,
|
|
lines,
|
|
level + 1
|
|
)
|
|
|
|
def resolve_url(self, base_url, href):
|
|
if not href:
|
|
return None
|
|
|
|
href = href.strip()
|
|
|
|
if href.startswith("#"):
|
|
return None
|
|
|
|
if href.startswith(("javascript:", "mailto:", "tel:")):
|
|
return None
|
|
|
|
if self.url_manager:
|
|
resolved = self.url_manager.resolve(base_url, href)
|
|
|
|
if not resolved:
|
|
return None
|
|
|
|
if not self.url_manager.is_allowed(resolved):
|
|
return None
|
|
|
|
return resolved
|
|
|
|
return urljoin(base_url, href)
|
|
|
|
def parse_sphinx_toctree(
|
|
self,
|
|
soup,
|
|
base_url
|
|
):
|
|
selectors = [
|
|
".toctree-wrapper",
|
|
".sphinxsidebar",
|
|
".sphinxsidebarwrapper",
|
|
".globaltoc",
|
|
".localtoc",
|
|
".contents.topic"
|
|
]
|
|
|
|
results = []
|
|
|
|
for selector in selectors:
|
|
for node in soup.select(selector):
|
|
tree = self.parse_list_structure(node, base_url)
|
|
|
|
if tree:
|
|
results.extend(tree)
|
|
|
|
return results
|
|
|
|
def dedupe_tree(self, tree):
|
|
seen = set()
|
|
result = []
|
|
|
|
for item in tree:
|
|
url = item.get("url")
|
|
|
|
if url and url in seen:
|
|
continue
|
|
|
|
if url:
|
|
seen.add(url)
|
|
|
|
children = item.get("children", [])
|
|
|
|
item["children"] = self.dedupe_tree(children)
|
|
|
|
result.append(item)
|
|
|
|
return result
|
|
|
|
def parse_breadcrumbs(
|
|
self,
|
|
soup,
|
|
base_url
|
|
):
|
|
selectors = [
|
|
".breadcrumb",
|
|
".breadcrumbs",
|
|
"nav[aria-label='breadcrumb']",
|
|
".wy-breadcrumbs",
|
|
".related"
|
|
]
|
|
|
|
items = []
|
|
|
|
for selector in selectors:
|
|
for node in soup.select(selector):
|
|
for a in node.find_all("a", href=True):
|
|
title = a.get_text(" ", strip=True)
|
|
href = a.get("href")
|
|
|
|
url = self.resolve_url(base_url, href)
|
|
|
|
if not url:
|
|
continue
|
|
|
|
items.append({
|
|
"title": title or url.rsplit("/", 1)[-1],
|
|
"url": url,
|
|
"children": []
|
|
})
|
|
|
|
return items |