327 lines
13 KiB
Python
327 lines
13 KiB
Python
"""OneNote loader utilities.
|
|
|
|
Provides load_OneNote_document which supports:
|
|
- Loading from local exported OneNote HTML directory or single .html/.mht file.
|
|
- Loading via Microsoft Graph API when provided with "graph" source and an access token.
|
|
|
|
This implementation keeps dependencies optional: only `requests` is required for Graph fetches.
|
|
"""
|
|
from typing import Optional, Dict, Any, List
|
|
import pathlib
|
|
import mimetypes
|
|
import logging
|
|
import re
|
|
import email
|
|
from email import policy
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Minimal contract:
|
|
# - input: path_or_source (str), options (dict)
|
|
# - output: Dict with keys: "type" ("local"|"graph"), "pages": list of {"title","content","id","lastModified"}
|
|
|
|
|
|
def _read_local_html_file(path: str) -> str:
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
# try to decode utf-8, fallback to latin1
|
|
for enc in ("utf-8", "utf-8-sig", "latin-1"):
|
|
try:
|
|
return data.decode(enc)
|
|
except Exception:
|
|
continue
|
|
return data.decode("latin-1", errors="ignore")
|
|
|
|
|
|
def _decode_part_payload(part) -> str:
|
|
# part is an email.message.Message
|
|
raw = part.get_payload(decode=True)
|
|
if raw is None:
|
|
return ""
|
|
ch = part.get_content_charset()
|
|
if ch:
|
|
try:
|
|
return raw.decode(ch, errors="replace")
|
|
except Exception:
|
|
pass
|
|
for enc in ("utf-8", "utf-8-sig", "latin-1"):
|
|
try:
|
|
return raw.decode(enc)
|
|
except Exception:
|
|
continue
|
|
return raw.decode("latin-1", errors="replace")
|
|
|
|
|
|
def _extract_title_from_html(html: str) -> str:
|
|
m = re.search(r"<title>(.*?)</title>", html, flags=re.I | re.S)
|
|
if m:
|
|
return m.group(1).strip()
|
|
# try to find first h1/h2 as fallback
|
|
m2 = re.search(r"<h1[^>]*>(.*?)</h1>", html, flags=re.I | re.S)
|
|
if m2:
|
|
return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", m2.group(1))).strip()
|
|
return "(untitled)"
|
|
|
|
|
|
def _split_html_into_pages(html: str) -> List[str]:
|
|
"""Heuristic split of a single HTML that may contain multiple OneNote pages.
|
|
|
|
Tries multiple strategies in order and returns the first that yields more than one fragment:
|
|
1. class-based: <div class="...page...">
|
|
2. id-based: <div id="Page...">
|
|
3. header-based: split by repeated <h1> (or <h2>) elements which often mark page titles
|
|
|
|
Returns list of HTML fragments (each a standalone-ish HTML string) or single-element list if heuristics fail.
|
|
"""
|
|
# 1) class-based (existing)
|
|
pattern_class = re.compile(r"<div[^>]+class=[\"'][^\"']*(?:page|onenote-page|onenotepage|oneNotePage)[^\"']*[\"'][^>]*>", flags=re.I)
|
|
matches = list(pattern_class.finditer(html))
|
|
if len(matches) > 1:
|
|
parts = []
|
|
head_match = re.search(r"<head>(.*?)</head>", html, flags=re.I | re.S)
|
|
head_html = head_match.group(0) if head_match else ""
|
|
for i, m in enumerate(matches):
|
|
start = m.start()
|
|
end = matches[i + 1].start() if i + 1 < len(matches) else len(html)
|
|
body_fragment = html[start:end]
|
|
frag = "<html>" + head_html + "<body>" + body_fragment + "</body></html>"
|
|
parts.append(frag)
|
|
return parts
|
|
|
|
# 2) id-based: look for divs whose id begins with 'Page' or 'page' or 'id="P' common OneNote exports
|
|
pattern_id = re.compile(r"<div[^>]+id=[\"']?(?:Page|page|P)_[^\s\"'>]*[\"']?[^>]*>", flags=re.I)
|
|
matches_id = list(pattern_id.finditer(html))
|
|
if len(matches_id) > 1:
|
|
parts = []
|
|
head_match = re.search(r"<head>(.*?)</head>", html, flags=re.I | re.S)
|
|
head_html = head_match.group(0) if head_match else ""
|
|
for i, m in enumerate(matches_id):
|
|
start = m.start()
|
|
end = matches_id[i + 1].start() if i + 1 < len(matches_id) else len(html)
|
|
body_fragment = html[start:end]
|
|
frag = "<html>" + head_html + "<body>" + body_fragment + "</body></html>"
|
|
parts.append(frag)
|
|
return parts
|
|
|
|
# 3) header-based: split by repeated <h1> (or <h2>) which often denote page titles in OneNote exports
|
|
h1_matches = list(re.finditer(r"<h1[^>]*>", html, flags=re.I))
|
|
if len(h1_matches) > 1:
|
|
parts = []
|
|
head_match = re.search(r"<head>(.*?)</head>", html, flags=re.I | re.S)
|
|
head_html = head_match.group(0) if head_match else ""
|
|
for i, m in enumerate(h1_matches):
|
|
start = m.start()
|
|
end = h1_matches[i + 1].start() if i + 1 < len(h1_matches) else len(html)
|
|
body_fragment = html[start:end]
|
|
frag = "<html>" + head_html + "<body>" + body_fragment + "</body></html>"
|
|
parts.append(frag)
|
|
return parts
|
|
|
|
# fallback: no multi-page split detected
|
|
return [html]
|
|
|
|
|
|
def _parse_mht_bytes(data: bytes) -> List[Dict[str, Any]]:
|
|
"""Parse MHT/MHTML bytes and return a list of pages (dicts with id/title/content).
|
|
|
|
Strategy:
|
|
- Use email.parser.BytesParser to parse the multipart MHT.
|
|
- For each part with content-type text/html, treat as one page.
|
|
- If there's only a single html part but it seems to contain multiple pages, try heuristic splitting.
|
|
"""
|
|
msg = email.parser.BytesParser(policy=policy.default).parsebytes(data)
|
|
html_parts: List[str] = []
|
|
ids: List[str] = []
|
|
# walk parts
|
|
if msg.is_multipart():
|
|
for part in msg.walk():
|
|
ctype = part.get_content_type()
|
|
if ctype == "text/html":
|
|
try:
|
|
html = _decode_part_payload(part)
|
|
except Exception:
|
|
html = _decode_part_payload(part)
|
|
html_parts.append(html)
|
|
# try to get a useful id from Content-Location or Content-ID
|
|
loc = part.get("Content-Location") or part.get("Content-ID") or ""
|
|
ids.append(loc)
|
|
else:
|
|
# not multipart: try parse as text/html
|
|
ctype = msg.get_content_type()
|
|
if ctype == "text/html":
|
|
html_parts.append(_decode_part_payload(msg))
|
|
ids.append("")
|
|
pages: List[Dict[str, Any]] = []
|
|
if len(html_parts) == 0:
|
|
return pages
|
|
if len(html_parts) == 1:
|
|
# maybe this single html contains multiple OneNote pages; try splitting heuristically
|
|
subpages = _split_html_into_pages(html_parts[0])
|
|
if len(subpages) > 1:
|
|
for idx, sp in enumerate(subpages):
|
|
title = _extract_title_from_html(sp) or f"page_{idx}"
|
|
pages.append({"id": ids[0] or f"mht_part_{idx}", "title": title, "content": sp})
|
|
return pages
|
|
# otherwise single page
|
|
title = _extract_title_from_html(html_parts[0])
|
|
pages.append({"id": ids[0] or "mht_root", "title": title, "content": html_parts[0]})
|
|
return pages
|
|
# multiple html parts -> each is a page
|
|
for idx, html in enumerate(html_parts):
|
|
title = _extract_title_from_html(html)
|
|
pages.append({"id": ids[idx] if idx < len(ids) else f"mht_part_{idx}", "title": title, "content": html})
|
|
return pages
|
|
|
|
|
|
def load_OneNote_document(path_or_source: str, options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Load OneNote content.
|
|
|
|
If `path_or_source` is a local filesystem path to a directory or file, load local HTML/MHT exports.
|
|
If `path_or_source` == "graph", options must contain `access_token` and either `notebook_id` or `site_id` + `notebook_id`.
|
|
|
|
Returns a dict with parsed pages.
|
|
"""
|
|
options = options or {}
|
|
|
|
p = pathlib.Path(path_or_source)
|
|
if path_or_source == "graph":
|
|
# fetch via Microsoft Graph
|
|
token = options.get("access_token")
|
|
if not token:
|
|
raise ValueError("Graph source requires 'access_token' in options")
|
|
notebook_id = options.get("notebook_id")
|
|
# simple graph pagination for pages
|
|
import requests
|
|
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
pages = []
|
|
if not notebook_id:
|
|
# try current user's notebooks
|
|
url = "https://graph.microsoft.com/v1.0/me/onenote/notebooks"
|
|
else:
|
|
url = f"https://graph.microsoft.com/v1.0/me/onenote/notebooks/{notebook_id}/sections"
|
|
resp = requests.get(url, headers=headers)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
# This is a thin implementation: collect pages by iterating sections then pages
|
|
sections = data.get("value", [])
|
|
for sec in sections:
|
|
sec_id = sec.get("id")
|
|
sec_title = sec.get("displayName")
|
|
# list pages in section
|
|
pages_url = f"https://graph.microsoft.com/v1.0/me/onenote/sections/{sec_id}/pages"
|
|
r2 = requests.get(pages_url, headers=headers)
|
|
if r2.status_code != 200:
|
|
continue
|
|
for pjson in r2.json().get("value", []):
|
|
content_url = pjson.get("contentUrl")
|
|
page_id = pjson.get("id")
|
|
title = pjson.get("title")
|
|
last_modified = pjson.get("lastModifiedDateTime")
|
|
# fetch content (HTML)
|
|
try:
|
|
c_resp = requests.get(content_url, headers=headers)
|
|
c_resp.raise_for_status()
|
|
content = c_resp.text
|
|
except Exception as e:
|
|
logger.exception("Failed to fetch OneNote page content: %s", e)
|
|
content = ""
|
|
pages.append({"id": page_id, "title": title, "content": content, "lastModified": last_modified, "section": sec_title})
|
|
return {"type": "graph", "pages": pages}
|
|
|
|
# Local path handling
|
|
if not p.exists():
|
|
raise FileNotFoundError(f"Path not found: {path_or_source}")
|
|
|
|
pages = []
|
|
if p.is_file():
|
|
mtype, _ = mimetypes.guess_type(p.as_posix())
|
|
if (p.suffix.lower() == ".mht") or (mtype == "message/rfc822") or (p.suffix.lower() == ".mhtml"):
|
|
# .mht parsing: try to parse parts and split into pages
|
|
with open(p.as_posix(), "rb") as f:
|
|
data = f.read()
|
|
try:
|
|
mht_pages = _parse_mht_bytes(data)
|
|
if mht_pages:
|
|
pages.extend(mht_pages)
|
|
else:
|
|
# fallback to raw read
|
|
content = _read_local_html_file(p.as_posix())
|
|
pages.append({"id": p.name, "title": p.stem, "content": content})
|
|
except Exception as e:
|
|
logger.exception("Failed to parse MHT file %s: %s", p, e)
|
|
content = _read_local_html_file(p.as_posix())
|
|
pages.append({"id": p.name, "title": p.stem, "content": content})
|
|
else:
|
|
content = _read_local_html_file(p.as_posix())
|
|
pages.append({"id": p.name, "title": p.stem, "content": content})
|
|
return {"type": "local", "pages": pages}
|
|
|
|
# directory: search for .html/.htm/.mht/.mhtml
|
|
exts = (".html", ".htm", ".mht", ".mhtml")
|
|
for fp in sorted(p.rglob("*")):
|
|
if fp.suffix.lower() in exts and fp.is_file():
|
|
try:
|
|
if fp.suffix.lower() in (".mht", ".mhtml"):
|
|
with open(fp.as_posix(), "rb") as f:
|
|
data = f.read()
|
|
mht_pages = _parse_mht_bytes(data)
|
|
if mht_pages:
|
|
for mp in mht_pages:
|
|
pages.append({"id": fp.as_posix() + "::" + mp.get("id", ""), "title": mp.get("title"), "content": mp.get("content")})
|
|
continue
|
|
# else fall through to raw read
|
|
content = _read_local_html_file(fp.as_posix())
|
|
except Exception as e:
|
|
logger.exception("Failed to read file %s: %s", fp, e)
|
|
continue
|
|
# try to extract title from HTML
|
|
m = re.search(r"<title>(.*?)</title>", content, flags=re.I | re.S)
|
|
title = m.group(1).strip() if m else fp.stem
|
|
pages.append({"id": fp.as_posix(), "title": title, "content": content})
|
|
return {"type": "local", "pages": pages}
|
|
|
|
|
|
def _safe_filename(s: str) -> str:
|
|
# create a filesystem-safe filename
|
|
s = s or "page"
|
|
s = re.sub(r"[\\\/\:\*\?\"<>\|]", "_", s)
|
|
s = re.sub(r"\s+", "_", s)
|
|
return s[:180]
|
|
|
|
|
|
def _maybe_debug_save(pages: List[Dict[str, Any]], options: Optional[Dict[str, Any]] = None) -> None:
|
|
"""If options contains 'debug_save_dir', save each page's HTML into that directory for inspection."""
|
|
options = options or {}
|
|
outdir = options.get("debug_save_dir")
|
|
if not outdir:
|
|
return
|
|
outp = pathlib.Path(outdir)
|
|
try:
|
|
outp.mkdir(parents=True, exist_ok=True)
|
|
except Exception:
|
|
return
|
|
for idx, page in enumerate(pages):
|
|
try:
|
|
title = page.get("title") or f"page_{idx}"
|
|
fname = f"{idx:03d}_{_safe_filename(title)}.html"
|
|
fp = outp.joinpath(fname)
|
|
with open(fp, "w", encoding="utf-8") as f:
|
|
f.write(page.get("content") or "")
|
|
except Exception:
|
|
logger.exception("Failed to write debug page %s", page.get("id"))
|
|
|
|
|
|
# Small smoke-run when executed directly
|
|
if __name__ == "__main__":
|
|
import sys
|
|
if len(sys.argv) > 1:
|
|
path = sys.argv[1]
|
|
else:
|
|
path = r'C:\Users\zhouyr9\Documents\2025.mht'
|
|
|
|
with open(path, "r", encoding='latin-1') as f:
|
|
content = f.read()
|
|
out = load_OneNote_document(path)
|
|
print(f"Found {len(out['pages'])} pages")
|