1470 lines
59 KiB
Python
1470 lines
59 KiB
Python
import io
|
||
import re
|
||
import sys
|
||
import html
|
||
import json
|
||
import zipfile
|
||
import posixpath
|
||
import urllib.parse
|
||
import urllib.request
|
||
import urllib.error
|
||
from pathlib import Path
|
||
import xml.etree.ElementTree as ET
|
||
from html.parser import HTMLParser
|
||
from typing import Tuple, List, Dict, Optional
|
||
|
||
|
||
def count_words(text: str) -> int:
|
||
"""
|
||
Counts words in text across multiple languages, ignoring isolated punctuation.
|
||
"""
|
||
if not text:
|
||
return 0
|
||
return len(re.findall(r"\b[\w]+(?:['’][\w]+)?\b", text))
|
||
|
||
|
||
class HTMLTextExtractor(HTMLParser):
|
||
"""
|
||
Strips HTML tags cleanly while preserving paragraph and heading breaks with newlines.
|
||
Also extracts the first heading and/or <title> tag for chapter naming.
|
||
"""
|
||
def __init__(self):
|
||
super().__init__()
|
||
self.text_parts: List[str] = []
|
||
self.ignore_tags = {"style", "script", "noscript"}
|
||
self.ignore_depth = 0
|
||
self.heading: str | None = None
|
||
self.in_heading = False
|
||
self.heading_parts: List[str] = []
|
||
self.title: str | None = None
|
||
self.in_title = False
|
||
self.title_parts: List[str] = []
|
||
|
||
def handle_starttag(self, tag: str, attrs):
|
||
t = tag.lower()
|
||
if t in self.ignore_tags:
|
||
self.ignore_depth += 1
|
||
elif t == "title":
|
||
self.in_title = True
|
||
self.title_parts = []
|
||
elif t in {"h1", "h2", "h3", "h4", "h5", "h6"}:
|
||
self.text_parts.append("\n\n")
|
||
if not self.heading:
|
||
self.in_heading = True
|
||
self.heading_parts = []
|
||
elif t in {"p", "div", "section", "article", "blockquote", "header", "footer"}:
|
||
self.text_parts.append("\n\n")
|
||
elif t in {"br", "hr", "li", "tr"}:
|
||
self.text_parts.append("\n")
|
||
|
||
def handle_endtag(self, tag: str):
|
||
t = tag.lower()
|
||
if t in self.ignore_tags:
|
||
self.ignore_depth = max(0, self.ignore_depth - 1)
|
||
elif t == "title":
|
||
self.in_title = False
|
||
self.title = "".join(self.title_parts).strip()
|
||
elif t in {"h1", "h2", "h3", "h4", "h5", "h6"}:
|
||
self.text_parts.append("\n\n")
|
||
if self.in_heading:
|
||
self.in_heading = False
|
||
if not self.heading:
|
||
self.heading = "".join(self.heading_parts).strip()
|
||
elif t in {"p", "div", "section", "article", "blockquote", "header", "footer"}:
|
||
self.text_parts.append("\n\n")
|
||
elif t in {"li", "tr"}:
|
||
self.text_parts.append("\n")
|
||
|
||
def handle_data(self, data: str):
|
||
if self.in_title:
|
||
self.title_parts.append(data)
|
||
if self.ignore_depth == 0 and not self.in_title:
|
||
# Replace non-breaking spaces with standard space
|
||
cleaned_data = data.replace("\xa0", " ")
|
||
self.text_parts.append(cleaned_data)
|
||
if self.in_heading:
|
||
self.heading_parts.append(cleaned_data)
|
||
|
||
def get_text(self) -> str:
|
||
raw = "".join(self.text_parts)
|
||
lines = [line.strip() for line in raw.split("\n")]
|
||
clean = "\n".join(lines)
|
||
return re.sub(r"\n{3,}", "\n\n", clean).strip()
|
||
|
||
|
||
def parse_epub(file_bytes: bytes) -> Tuple[str, str, List[Dict[str, str]]]:
|
||
"""
|
||
Parses an EPUB file archive and extracts:
|
||
- title: book title
|
||
- author: creator/author
|
||
- chapters: list of {"title": str, "content": str}
|
||
"""
|
||
try:
|
||
z = zipfile.ZipFile(io.BytesIO(file_bytes))
|
||
except Exception as e:
|
||
raise ValueError(f"Invalid EPUB archive: {e}")
|
||
|
||
# 1. Locate root OPF file from META-INF/container.xml
|
||
opf_path = None
|
||
try:
|
||
container_data = z.read("META-INF/container.xml")
|
||
root = ET.fromstring(container_data)
|
||
rootfile = root.find(".//{*}rootfile")
|
||
if rootfile is not None:
|
||
opf_path = rootfile.attrib.get("full-path")
|
||
except Exception:
|
||
pass
|
||
|
||
if not opf_path:
|
||
for name in z.namelist():
|
||
if name.lower().endswith(".opf"):
|
||
opf_path = name
|
||
break
|
||
|
||
if not opf_path:
|
||
raise ValueError("Invalid EPUB: no OPF file found")
|
||
|
||
opf_path = opf_path.replace("\\", "/").lstrip("/")
|
||
opf_dir = posixpath.dirname(opf_path)
|
||
|
||
try:
|
||
opf_content = z.read(opf_path)
|
||
except KeyError:
|
||
# Try lowercase matching
|
||
name_map = {n.lower(): n for n in z.namelist()}
|
||
if opf_path.lower() in name_map:
|
||
opf_content = z.read(name_map[opf_path.lower()])
|
||
else:
|
||
raise ValueError(f"OPF file not found in archive: {opf_path}")
|
||
|
||
# 2. Extract metadata (title, author)
|
||
title = "Untitled"
|
||
author = ""
|
||
opf_tree = None
|
||
try:
|
||
opf_tree = ET.fromstring(opf_content)
|
||
title_el = opf_tree.find(".//{*}title")
|
||
if title_el is not None and title_el.text:
|
||
title = title_el.text.strip()
|
||
creator_el = opf_tree.find(".//{*}creator")
|
||
if creator_el is not None and creator_el.text:
|
||
author = creator_el.text.strip()
|
||
except Exception:
|
||
opf_str = opf_content.decode("utf-8", errors="ignore")
|
||
tm = re.search(r"<[^:]*:?title[^>]*>(.*?)</[^:]*:?title>", opf_str, re.I | re.S)
|
||
if tm:
|
||
title = re.sub(r"<[^>]+>", "", tm.group(1)).strip()
|
||
cm = re.search(r"<[^:]*:?creator[^>]*>(.*?)</[^:]*:?creator>", opf_str, re.I | re.S)
|
||
if cm:
|
||
author = re.sub(r"<[^>]+>", "", cm.group(1)).strip()
|
||
|
||
# 3. Parse Manifest (id -> href, media-type)
|
||
manifest: Dict[str, Dict[str, str]] = {}
|
||
if opf_tree is not None:
|
||
for item in opf_tree.findall(".//{*}item"):
|
||
iid = item.attrib.get("id")
|
||
href = item.attrib.get("href")
|
||
media_type = item.attrib.get("media-type", "").lower()
|
||
if iid and href:
|
||
manifest[iid] = {"href": href, "media_type": media_type}
|
||
else:
|
||
opf_str = opf_content.decode("utf-8", errors="ignore")
|
||
for m in re.finditer(r'<item\s+[^>]*id=["\']([^"\']+)["\'][^>]*href=["\']([^"\']+)["\']', opf_str, re.I):
|
||
manifest[m.group(1)] = {"href": m.group(2), "media_type": ""}
|
||
|
||
# 4. Spine items in reading order
|
||
spine_ids: List[str] = []
|
||
if opf_tree is not None:
|
||
for itemref in opf_tree.findall(".//{*}itemref"):
|
||
idref = itemref.attrib.get("idref")
|
||
if idref:
|
||
spine_ids.append(idref)
|
||
else:
|
||
opf_str = opf_content.decode("utf-8", errors="ignore")
|
||
for m in re.finditer(r'<itemref\s+[^>]*idref=["\']([^"\']+)["\']', opf_str, re.I):
|
||
spine_ids.append(m.group(1))
|
||
|
||
# Fallback if spine is empty: find all html/xhtml files in manifest
|
||
candidate_paths: List[str] = []
|
||
if spine_ids:
|
||
for idref in spine_ids:
|
||
item = manifest.get(idref)
|
||
if item:
|
||
candidate_paths.append(item["href"])
|
||
else:
|
||
for item in manifest.values():
|
||
href = item["href"]
|
||
if href.lower().endswith((".xhtml", ".html", ".htm")):
|
||
candidate_paths.append(href)
|
||
|
||
namelist_map = {name.lower(): name for name in z.namelist()}
|
||
chapters: List[Dict[str, str]] = []
|
||
chapter_num = 1
|
||
|
||
for href in candidate_paths:
|
||
unquoted = urllib.parse.unquote(href)
|
||
rel_path = posixpath.normpath(posixpath.join(opf_dir, unquoted)) if opf_dir else unquoted
|
||
rel_path = rel_path.replace("\\", "/").lstrip("/")
|
||
|
||
actual_path = None
|
||
if rel_path in z.namelist():
|
||
actual_path = rel_path
|
||
elif rel_path.lower() in namelist_map:
|
||
actual_path = namelist_map[rel_path.lower()]
|
||
|
||
if not actual_path:
|
||
continue
|
||
|
||
try:
|
||
raw_bytes = z.read(actual_path)
|
||
try:
|
||
html_str = raw_bytes.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
html_str = raw_bytes.decode("latin-1", errors="replace")
|
||
except Exception:
|
||
continue
|
||
|
||
parser = HTMLTextExtractor()
|
||
parser.feed(html_str)
|
||
text = parser.get_text()
|
||
|
||
# Skip chapters with little or no readable text
|
||
if count_words(text) < 1:
|
||
continue
|
||
|
||
ch_title = parser.heading or parser.title or f"Chapter {chapter_num}"
|
||
# If ch_title is identical to book title or very long, format nicely
|
||
if len(ch_title) > 120:
|
||
ch_title = f"Chapter {chapter_num}"
|
||
|
||
chapters.append({"title": ch_title, "content": text})
|
||
chapter_num += 1
|
||
|
||
if not chapters:
|
||
chapters = [{"title": title if title != "Untitled" else "Chapter 1", "content": ""}]
|
||
|
||
return title, author, chapters
|
||
|
||
|
||
def split_long_text(text: str, target_words: int = 3000) -> List[str]:
|
||
"""
|
||
Splits text into chunks of approximately target_words on paragraph boundaries.
|
||
"""
|
||
paragraphs = [p.strip() for p in re.split(r"\n{2,}", text) if p.strip()]
|
||
if not paragraphs:
|
||
return [text.strip()] if text.strip() else []
|
||
|
||
chunks: List[str] = []
|
||
current_paras: List[str] = []
|
||
current_count = 0
|
||
|
||
for p in paragraphs:
|
||
w_count = count_words(p)
|
||
# If single paragraph is massive (> target_words * 1.5), split by words
|
||
if w_count > target_words * 1.5:
|
||
if current_paras:
|
||
chunks.append("\n\n".join(current_paras))
|
||
current_paras = []
|
||
current_count = 0
|
||
words = p.split()
|
||
for i in range(0, len(words), target_words):
|
||
chunk = " ".join(words[i : i + target_words])
|
||
if chunk.strip():
|
||
chunks.append(chunk)
|
||
continue
|
||
|
||
if current_count + w_count > target_words and current_count > 0:
|
||
chunks.append("\n\n".join(current_paras))
|
||
current_paras = [p]
|
||
current_count = w_count
|
||
else:
|
||
current_paras.append(p)
|
||
current_count += w_count
|
||
|
||
if current_paras:
|
||
chunks.append("\n\n".join(current_paras))
|
||
|
||
return chunks
|
||
|
||
|
||
def parse_text(content: str, title: str = "Untitled", author: str = "") -> Tuple[str, str, List[Dict[str, str]]]:
|
||
"""
|
||
Parses plain text or Markdown into readable chapters.
|
||
Splits by explicit chapter markers (e.g. 'Chapter X', '# Chapter') or chunks of ~3000 words.
|
||
"""
|
||
content = content.replace("\r\n", "\n").replace("\r", "\n").strip()
|
||
if not content:
|
||
return title, author, [{"title": title if title != "Untitled" else "Chapter 1", "content": ""}]
|
||
|
||
lines = content.split("\n")
|
||
first_line = lines[0].strip()
|
||
|
||
# Extract title from first line or # Heading / <h1...> if default or generic README
|
||
if title == "Untitled" or title.lower().startswith("readme"):
|
||
if first_line.startswith("# "):
|
||
title = first_line.lstrip("# ").strip()
|
||
if len(lines) > 1:
|
||
content = "\n".join(lines[1:]).strip()
|
||
elif first_line.lower().startswith("title:"):
|
||
title = first_line.split(":", 1)[1].strip()
|
||
if len(lines) > 1:
|
||
content = "\n".join(lines[1:]).strip()
|
||
else:
|
||
h1_match = re.search(r"<h1[^>]*>(.*?)</h1>", content[:1500], re.IGNORECASE)
|
||
if h1_match:
|
||
detected = re.sub(r"<[^>]+>", "", h1_match.group(1)).strip()
|
||
if detected:
|
||
title = detected
|
||
|
||
# Extract author from next line if missing
|
||
if not author:
|
||
lines = content.split("\n")
|
||
if lines and lines[0].lower().startswith(("author:", "by:")):
|
||
author = lines[0].split(":", 1)[1].strip()
|
||
if len(lines) > 1:
|
||
content = "\n".join(lines[1:]).strip()
|
||
|
||
def detect_text_headings(content: str) -> List[Dict]:
|
||
"""
|
||
Detects section and chapter headings with high accuracy across Markdown,
|
||
numbered sections, explicit keywords, bold titles, and Setext styles.
|
||
"""
|
||
lines = content.split("\n")
|
||
line_starts = []
|
||
curr = 0
|
||
for l in lines:
|
||
line_starts.append(curr)
|
||
curr += len(l) + 1
|
||
|
||
kw_regex = re.compile(
|
||
r"^[ \t]*(?:\*\*)?(?:Chapter|CHAPTER|Глава|ГЛАВА|Part|PART|Часть|ЧАСТЬ|Section|SECTION|Раздел|РАЗДЕЛ|Module|MODULE|Модуль|МОДУЛЬ|Unit|UNIT|Lesson|LESSON|Урок|УРОК|Topic|TOPIC|Тема|ТЕМА|Appendix|APPENDIX|Приложение|ПРИЛОЖЕНИЕ|Act|ACT|Scene|SCENE)[ \t]+(?:[0-9IVXLCDMivxlcdm\w\.\-]+)(?:[ \t]*[:\-–—\.][ \t]*[^\n\r*]+|[ \t]+[^\n\r*]+)?(?:\*\*)?[ \t]*$"
|
||
)
|
||
|
||
headings = []
|
||
|
||
for i, line in enumerate(lines):
|
||
trimmed = line.strip()
|
||
if not trimmed:
|
||
continue
|
||
|
||
prev_empty = (i == 0) or (not lines[i - 1].strip())
|
||
next_empty = (i == len(lines) - 1) or (not lines[i + 1].strip())
|
||
|
||
# 1. Markdown # heading (# to #####)
|
||
m_md = re.match(r"^#{1,5}[ \t]+(.+)$", trimmed)
|
||
if m_md:
|
||
clean_title = m_md.group(1).strip("#").strip()
|
||
headings.append({
|
||
"line_idx": i,
|
||
"start": line_starts[i],
|
||
"title": clean_title,
|
||
"type": "markdown",
|
||
})
|
||
continue
|
||
|
||
# 2. Setext heading: Title followed by === or ---
|
||
if i + 1 < len(lines):
|
||
next_line = lines[i + 1].strip()
|
||
if prev_empty and len(trimmed) <= 80 and not trimmed.startswith(("-", "*", "#", "`")):
|
||
if re.match(r"^={3,}$", next_line) or re.match(r"^-{3,}$", next_line):
|
||
headings.append({
|
||
"line_idx": i,
|
||
"start": line_starts[i],
|
||
"title": trimmed,
|
||
"type": "setext",
|
||
})
|
||
continue
|
||
|
||
# 3. Explicit keywords (Chapter 1, Глава 2, Module 3, etc.)
|
||
if kw_regex.match(trimmed):
|
||
clean_title = trimmed.strip("*").strip()
|
||
headings.append({
|
||
"line_idx": i,
|
||
"start": line_starts[i],
|
||
"title": clean_title,
|
||
"type": "keyword",
|
||
})
|
||
continue
|
||
|
||
# 4. Bold heading on standalone line: **Title**
|
||
bold_m = re.match(r"^\*\*([A-ZА-ЯЁ0-9][^\n\r*]{2,70})\*\*$", trimmed)
|
||
if bold_m and prev_empty:
|
||
headings.append({
|
||
"line_idx": i,
|
||
"start": line_starts[i],
|
||
"title": bold_m.group(1).strip(),
|
||
"type": "bold",
|
||
})
|
||
continue
|
||
|
||
# 5. Numbered sections: e.g. 1. Title, 1.1 Title, 1. Title.
|
||
num_m = re.match(r"^([0-9]{1,2}(?:\.[0-9]{1,2})*\.?|[0-9]{1,2}\))[ \t]+([A-ZА-ЯЁ][^\n\r]{2,70})$", trimmed)
|
||
if num_m and prev_empty:
|
||
is_list = False
|
||
if i + 1 < len(lines):
|
||
next_t = lines[i + 1].strip()
|
||
if re.match(r"^[0-9]{1,2}(?:\.[0-9]{1,2})*\.?[ \t]+", next_t):
|
||
is_list = True
|
||
if not is_list and not trimmed.endswith((",", ";")):
|
||
headings.append({
|
||
"line_idx": i,
|
||
"start": line_starts[i],
|
||
"title": trimmed,
|
||
"type": "numbered",
|
||
})
|
||
continue
|
||
|
||
# 6. Standalone ALL CAPS line (e.g. INTRODUCTION TO NETWORKS)
|
||
next_is_allcaps = (i + 1 < len(lines)) and bool(lines[i + 1].strip()) and lines[i + 1].strip().isupper()
|
||
if prev_empty and not next_is_allcaps and 4 <= len(trimmed) <= 60 and not trimmed.endswith((",", ";")):
|
||
if re.match(r"^[A-ZА-ЯЁ0-9\s:\-–—\.]+$", trimmed):
|
||
letters = re.findall(r"[A-ZА-ЯЁ]", trimmed)
|
||
words = trimmed.split()
|
||
if len(letters) >= 4 and (len(words) >= 2 or len(trimmed) >= 6):
|
||
headings.append({
|
||
"line_idx": i,
|
||
"start": line_starts[i],
|
||
"title": trimmed,
|
||
"type": "allcaps",
|
||
})
|
||
continue
|
||
|
||
return headings
|
||
|
||
|
||
def parse_text(content: str, title: str = "Untitled", author: str = "") -> Tuple[str, str, List[Dict[str, str]]]:
|
||
"""
|
||
Parses plain text or Markdown into readable chapters.
|
||
Splits by intelligent heading detection (Markdown, numbered sections, keywords)
|
||
or chunks of ~1000 words.
|
||
"""
|
||
content = content.replace("\r\n", "\n").replace("\r", "\n").strip()
|
||
if not content:
|
||
return title, author, [{"title": title if title != "Untitled" else "Chapter 1", "content": ""}]
|
||
|
||
lines = content.split("\n")
|
||
first_line = lines[0].strip()
|
||
|
||
# Extract title from first line or # Heading / <h1...> if default or generic README
|
||
if title == "Untitled" or title.lower().startswith("readme"):
|
||
if first_line.startswith("# "):
|
||
title = first_line.lstrip("# ").strip()
|
||
if len(lines) > 1:
|
||
content = "\n".join(lines[1:]).strip()
|
||
elif first_line.lower().startswith("title:"):
|
||
title = first_line.split(":", 1)[1].strip()
|
||
if len(lines) > 1:
|
||
content = "\n".join(lines[1:]).strip()
|
||
else:
|
||
h1_match = re.search(r"<h1[^>]*>(.*?)</h1>", content[:1500], re.IGNORECASE)
|
||
if h1_match:
|
||
detected = re.sub(r"<[^>]+>", "", h1_match.group(1)).strip()
|
||
if detected:
|
||
title = detected
|
||
|
||
# Extract author from next line if missing
|
||
if not author:
|
||
lines = content.split("\n")
|
||
if lines and lines[0].lower().startswith(("author:", "by:")):
|
||
author = lines[0].split(":", 1)[1].strip()
|
||
if len(lines) > 1:
|
||
content = "\n".join(lines[1:]).strip()
|
||
|
||
detected_headings = detect_text_headings(content)
|
||
chapters: List[Dict[str, str]] = []
|
||
|
||
if detected_headings:
|
||
# Pre-chapter intro text
|
||
intro = content[: detected_headings[0]["start"]].strip()
|
||
if intro and count_words(intro) > 10:
|
||
is_ru = any(ord(c) > 127 for c in intro[:200])
|
||
chapters.append({"title": "Введение" if is_ru else "Introduction", "content": intro})
|
||
|
||
for i, h in enumerate(detected_headings):
|
||
start = h["start"]
|
||
end = detected_headings[i + 1]["start"] if i + 1 < len(detected_headings) else len(content)
|
||
body = content[start:end].strip()
|
||
if body:
|
||
chapters.append({"title": h["title"], "content": body})
|
||
|
||
if not chapters:
|
||
# No headings detected: split by ~1000 words chunks if long enough
|
||
total_w = count_words(content)
|
||
if total_w > 1200:
|
||
chunks = split_long_text(content, target_words=1000)
|
||
is_ru = any(ord(c) > 127 for c in content[:200])
|
||
prefix = "Часть" if is_ru else "Chapter"
|
||
for idx, chunk in enumerate(chunks, start=1):
|
||
chapters.append({"title": f"{prefix} {idx}", "content": chunk})
|
||
else:
|
||
chapters = [{"title": title if title != "Untitled" else "Chapter 1", "content": content}]
|
||
|
||
return title, author, chapters
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# PDF Parsing & Chapter Extraction (pypdf, Bookmarks/Outline, Cisco CLI Blocks)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
CLI_PROMPT_RE = re.compile(
|
||
r"^(?:[A-Za-z0-9_.\-]+(?:\([a-zA-Z0-9_\-\.\s]+\))?[>#]|"
|
||
r"[A-Za-z0-9_.\-]+@[A-Za-z0-9_.\-]+:[~/\w\.\-]*[$#]|"
|
||
r"(?:[A-Z]):\\[^>\n\r]*>|"
|
||
r"PS\s+[A-Z]:\\[^>\n\r]*>)"
|
||
)
|
||
|
||
|
||
def extract_outline_entries(outline, reader) -> List[Tuple[str, int]]:
|
||
"""
|
||
Recursively extracts bookmarks / outline entries from PDF respecting hierarchy depth.
|
||
For books with deep trees (Parts -> Chapters -> Sections -> Subsections),
|
||
prioritizes major structural headings and prevents fragmenting chapters into
|
||
hundreds of tiny 1-page snippets.
|
||
"""
|
||
raw_entries: List[Tuple[str, int, int]] = []
|
||
|
||
def _traverse(items, depth=0):
|
||
for item in items:
|
||
if isinstance(item, list):
|
||
_traverse(item, depth + 1)
|
||
else:
|
||
try:
|
||
title = getattr(item, "title", None) or (item.get("/Title") if hasattr(item, "get") else None)
|
||
page_num = reader.get_destination_page_number(item)
|
||
if title and page_num is not None:
|
||
clean_title = str(title).strip()
|
||
if clean_title:
|
||
raw_entries.append((clean_title, int(page_num), depth))
|
||
except Exception:
|
||
continue
|
||
|
||
if outline:
|
||
_traverse(outline, 0)
|
||
|
||
if not raw_entries:
|
||
return []
|
||
|
||
# If total entries is reasonable (<= 50), return all
|
||
if len(raw_entries) <= 50:
|
||
return [(t, p) for t, p, _ in raw_entries]
|
||
|
||
# Check depth distribution
|
||
d0_entries = [e for e in raw_entries if e[2] == 0]
|
||
d1_entries = [e for e in raw_entries if e[2] <= 1]
|
||
|
||
# If top-level (depth 0) already has >= 10 chapters, use depth 0
|
||
if len(d0_entries) >= 10:
|
||
return [(t, p) for t, p, _ in d0_entries]
|
||
|
||
# Otherwise, if depth <= 1 gives a good number of chapters (10 to 80), use depth <= 1
|
||
if len(d1_entries) <= 80:
|
||
return [(t, p) for t, p, _ in d1_entries]
|
||
|
||
# If still too many entries (> 80), filter by major chapter keywords
|
||
major_re = re.compile(
|
||
r"^(cover|title\s+page|copyright|contents|introduction|your\s+study\s+plan|"
|
||
r"part\s+[0-9ivxlcdm]+|chapter\s+\d+|appendix\s+[a-z]|glossary|index)",
|
||
re.IGNORECASE
|
||
)
|
||
major_entries = [e for e in d1_entries if major_re.match(e[0])]
|
||
if len(major_entries) >= 10:
|
||
return [(t, p) for t, p, _ in major_entries]
|
||
|
||
return [(t, p) for t, p, _ in d1_entries[:80]]
|
||
|
||
|
||
|
||
def clean_page_lines(raw_pages_text: List[str]) -> List[str]:
|
||
"""
|
||
Filters running headers, footers, isolated page numbers, and copyright noise.
|
||
"""
|
||
if not raw_pages_text:
|
||
return []
|
||
|
||
num_pages = len(raw_pages_text)
|
||
top_candidates: Dict[str, int] = {}
|
||
bottom_candidates: Dict[str, int] = {}
|
||
|
||
for pt in raw_pages_text:
|
||
lines = [l.strip() for l in pt.split("\n") if l.strip()]
|
||
if not lines:
|
||
continue
|
||
if len(lines) >= 1:
|
||
top_candidates[lines[0]] = top_candidates.get(lines[0], 0) + 1
|
||
bottom_candidates[lines[-1]] = bottom_candidates.get(lines[-1], 0) + 1
|
||
|
||
# Lines repeated on >= 20% of pages (min 3 pages) are running headers/footers
|
||
threshold = max(3, int(num_pages * 0.20))
|
||
common_headers = {l for l, c in top_candidates.items() if c >= threshold}
|
||
common_footers = {l for l, c in bottom_candidates.items() if c >= threshold}
|
||
|
||
page_num_re = re.compile(r"^(?:page\s+)?\d+(?:\s*(?:of|/)\s*\d+)?$", re.IGNORECASE)
|
||
url_re = re.compile(r"^(?:https?://|www\.)\S+$", re.IGNORECASE)
|
||
copyright_re = re.compile(r"^(?:©|\(c\)|copyright)\s*\d{4}.*$", re.IGNORECASE)
|
||
|
||
cleaned_pages: List[str] = []
|
||
for pt in raw_pages_text:
|
||
lines = pt.split("\n")
|
||
filtered: List[str] = []
|
||
n_lines = len(lines)
|
||
for idx, l in enumerate(lines):
|
||
stripped = l.strip()
|
||
if not stripped:
|
||
filtered.append("")
|
||
continue
|
||
is_top = (idx <= 2)
|
||
is_bottom = (idx >= n_lines - 3)
|
||
|
||
if (is_top or is_bottom) and page_num_re.match(stripped):
|
||
continue
|
||
if is_top and stripped in common_headers:
|
||
continue
|
||
if is_bottom and (stripped in common_footers or url_re.match(stripped) or copyright_re.match(stripped)):
|
||
continue
|
||
filtered.append(l)
|
||
cleaned_pages.append("\n".join(filtered))
|
||
|
||
return cleaned_pages
|
||
|
||
|
||
def format_cli_blocks(text: str) -> str:
|
||
"""
|
||
Detects Cisco IOS / CLI terminal sessions in text and wraps them
|
||
in Markdown fenced code blocks (```text ... ```) for syntax highlighting and copying.
|
||
"""
|
||
if not text:
|
||
return ""
|
||
|
||
lines = text.split("\n")
|
||
out: List[str] = []
|
||
in_cli = False
|
||
cli_lines: List[str] = []
|
||
|
||
def is_likely_cli_line(line: str) -> bool:
|
||
s = line.strip()
|
||
if not s:
|
||
return True
|
||
if CLI_PROMPT_RE.match(s):
|
||
return True
|
||
if s.startswith(("%", "!", "--", ">>", "Enter configuration", "Building configuration", "Current configuration")):
|
||
return True
|
||
if len(re.findall(r"\S\s{2,}\S", line)) >= 2:
|
||
return True
|
||
if re.match(r"^(?:show|ip|no|interface|switchport|vlan|router|network|exit|end|enable|disable|configure|ping|traceroute|debug)\s+", s, re.IGNORECASE):
|
||
return True
|
||
return False
|
||
|
||
for line in lines:
|
||
s = line.strip()
|
||
if not in_cli:
|
||
if CLI_PROMPT_RE.match(s):
|
||
in_cli = True
|
||
cli_lines = [line]
|
||
else:
|
||
out.append(line)
|
||
else:
|
||
if is_likely_cli_line(line):
|
||
cli_lines.append(line)
|
||
else:
|
||
while cli_lines and not cli_lines[-1].strip():
|
||
cli_lines.pop()
|
||
if cli_lines:
|
||
out.append("```text")
|
||
out.extend(cli_lines)
|
||
out.append("```")
|
||
in_cli = False
|
||
cli_lines = []
|
||
out.append(line)
|
||
|
||
if in_cli and cli_lines:
|
||
while cli_lines and not cli_lines[-1].strip():
|
||
cli_lines.pop()
|
||
if cli_lines:
|
||
out.append("```text")
|
||
out.extend(cli_lines)
|
||
out.append("```")
|
||
|
||
return "\n".join(out)
|
||
|
||
|
||
def parse_pdf(file_bytes: bytes, filename: str = "") -> Tuple[str, str, List[Dict[str, str]]]:
|
||
"""
|
||
Parses a PDF file into readable chapters:
|
||
- Extracts title and author from metadata or filename.
|
||
- Enforces Scanned PDF Guard (raises ValueError if total_words < 100).
|
||
- Cleans repetitive headers, footers, and page numbers.
|
||
- Wraps Cisco IOS / CLI terminal blocks in Markdown code blocks.
|
||
- Splits chapters by PDF outline / bookmarks (Cisco Press, O'Reilly).
|
||
- Fallback to heading regex or ~10-15 page chunks if no bookmarks exist.
|
||
"""
|
||
try:
|
||
import pypdf
|
||
except ImportError:
|
||
vendor_dir = Path(__file__).resolve().parent / "vendor"
|
||
if str(vendor_dir) not in sys.path:
|
||
sys.path.insert(0, str(vendor_dir))
|
||
import pypdf
|
||
|
||
try:
|
||
reader = pypdf.PdfReader(io.BytesIO(file_bytes))
|
||
except Exception as e:
|
||
raise ValueError(f"Не удалось открыть PDF документ: {e}")
|
||
|
||
if reader.is_encrypted:
|
||
try:
|
||
reader.decrypt("")
|
||
except Exception:
|
||
raise ValueError("Этот PDF документ защищен паролем. Пожалуйста, снимите пароль для чтения.")
|
||
|
||
num_pages = len(reader.pages)
|
||
if num_pages == 0:
|
||
raise ValueError("PDF документ не содержит страниц")
|
||
|
||
# Metadata extraction
|
||
meta = reader.metadata or {}
|
||
meta_title = getattr(meta, "title", None) or (meta.get("/Title") if hasattr(meta, "get") else None) or ""
|
||
meta_author = getattr(meta, "author", None) or (meta.get("/Author") if hasattr(meta, "get") else None) or ""
|
||
|
||
stem = Path(filename).stem if filename else "Untitled Document"
|
||
clean_title = str(meta_title).strip()
|
||
if not clean_title or clean_title.lower() in {"untitled", "untitled document", "default", "temp", "output"}:
|
||
title = stem
|
||
else:
|
||
title = clean_title
|
||
|
||
author = str(meta_author).strip()
|
||
|
||
# Page-by-page text extraction
|
||
raw_pages_text: List[str] = []
|
||
for p in reader.pages:
|
||
try:
|
||
pt = p.extract_text() or ""
|
||
except Exception:
|
||
pt = ""
|
||
raw_pages_text.append(pt)
|
||
|
||
# Scanned PDF Guard (< 100 words)
|
||
total_words = sum(count_words(t) for t in raw_pages_text)
|
||
if total_words < 100:
|
||
raise ValueError(
|
||
"Этот PDF-документ содержит сканы или изображения без текстового слоя (OCR). "
|
||
"Пожалуйста, используйте документ с распознанным текстом."
|
||
)
|
||
|
||
# Clean noise lines (headers, footers, page numbers)
|
||
cleaned_pages = clean_page_lines(raw_pages_text)
|
||
|
||
# Apply CLI formatting to each page
|
||
formatted_pages = [format_cli_blocks(p) for p in cleaned_pages]
|
||
|
||
# Chapter splitting:
|
||
# 1. Try outline / bookmarks
|
||
outline_entries = extract_outline_entries(getattr(reader, "outline", None), reader)
|
||
chapters: List[Dict[str, str]] = []
|
||
|
||
if outline_entries:
|
||
valid_entries = [(t, p) for t, p in outline_entries if 0 <= p < num_pages]
|
||
valid_entries.sort(key=lambda x: x[1])
|
||
|
||
# Filter duplicates pointing to the same page
|
||
deduped_entries: List[Tuple[str, int]] = []
|
||
for t, p in valid_entries:
|
||
if not deduped_entries or deduped_entries[-1][1] != p:
|
||
deduped_entries.append((t, p))
|
||
|
||
if deduped_entries:
|
||
first_p = deduped_entries[0][1]
|
||
if first_p > 0:
|
||
intro_content = "\n\n".join(formatted_pages[p].strip() for p in range(0, first_p) if formatted_pages[p].strip())
|
||
if count_words(intro_content) >= 20:
|
||
is_ru = any(ord(c) > 127 for c in intro_content[:300])
|
||
chapters.append({
|
||
"title": "Введение" if is_ru else "Introduction",
|
||
"content": intro_content
|
||
})
|
||
|
||
for i, (entry_title, start_p) in enumerate(deduped_entries):
|
||
end_p = deduped_entries[i + 1][1] if i + 1 < len(deduped_entries) else num_pages
|
||
ch_content = "\n\n".join(formatted_pages[p].strip() for p in range(start_p, end_p) if formatted_pages[p].strip())
|
||
if ch_content:
|
||
chapters.append({
|
||
"title": entry_title,
|
||
"content": ch_content
|
||
})
|
||
|
||
# 2. Fallback: No outline found, check for page-level chapter headings
|
||
if not chapters:
|
||
chapter_heading_re = re.compile(
|
||
r"^\s*(?:chapter|глава|part|часть|section|раздел|module|модуль)\s+([0-9ivxlcdm]+)[\s:.\-–—\n]+(.*)$",
|
||
re.IGNORECASE
|
||
)
|
||
detected_pages: List[Tuple[str, int]] = []
|
||
for p_idx, p_text in enumerate(formatted_pages):
|
||
first_lines = [l.strip() for l in p_text.split("\n") if l.strip()][:3]
|
||
for l in first_lines:
|
||
m = chapter_heading_re.match(l)
|
||
if m:
|
||
h_title = l.strip("#").strip()
|
||
detected_pages.append((h_title, p_idx))
|
||
break
|
||
|
||
if detected_pages:
|
||
if detected_pages[0][1] > 0:
|
||
intro_content = "\n\n".join(formatted_pages[p].strip() for p in range(0, detected_pages[0][1]) if formatted_pages[p].strip())
|
||
if count_words(intro_content) >= 20:
|
||
is_ru = any(ord(c) > 127 for c in intro_content[:300])
|
||
chapters.append({
|
||
"title": "Введение" if is_ru else "Introduction",
|
||
"content": intro_content
|
||
})
|
||
|
||
for i, (h_title, start_p) in enumerate(detected_pages):
|
||
end_p = detected_pages[i + 1][1] if i + 1 < len(detected_pages) else num_pages
|
||
ch_content = "\n\n".join(formatted_pages[p].strip() for p in range(start_p, end_p) if formatted_pages[p].strip())
|
||
if ch_content:
|
||
chapters.append({
|
||
"title": h_title,
|
||
"content": ch_content
|
||
})
|
||
|
||
# 3. Fallback: Group pages in chunks of 12-15 pages
|
||
if not chapters:
|
||
chunk_size = 12
|
||
sample_text = "".join(formatted_pages[:3])
|
||
is_ru = any(ord(c) > 127 for c in sample_text[:300])
|
||
prefix = "Страницы" if is_ru else "Pages"
|
||
|
||
for start_p in range(0, num_pages, chunk_size):
|
||
end_p = min(start_p + chunk_size, num_pages)
|
||
ch_content = "\n\n".join(formatted_pages[p].strip() for p in range(start_p, end_p) if formatted_pages[p].strip())
|
||
if ch_content:
|
||
chapters.append({
|
||
"title": f"{prefix} {start_p + 1}–{end_p}",
|
||
"content": ch_content
|
||
})
|
||
|
||
if not chapters:
|
||
full_content = "\n\n".join(p.strip() for p in formatted_pages if p.strip())
|
||
chapters = [{"title": title, "content": full_content}]
|
||
|
||
return title, author, chapters
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# RFC Specifications Parser & Fetcher (IETF & RFC Editor)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def extract_rfc_number(query: str) -> int:
|
||
"""
|
||
Extracts RFC number from a string, supporting inputs like:
|
||
- "9110", "rfc 9110", "RFC-9110", "rfc9110"
|
||
- "https://www.rfc-editor.org/rfc/rfc9110.txt"
|
||
- "https://datatracker.ietf.org/doc/rfc9110/"
|
||
"""
|
||
query = (query or "").strip()
|
||
if re.search(r"-\s*\d+", query) and not re.search(r"rfc-\d+", query, re.IGNORECASE):
|
||
raise ValueError(f"Недопустимый отрицательный номер RFC: '{query}'")
|
||
|
||
m = re.search(r"(?:rfc[-_ ]?|doc/(?:html/)?rfc|/rfc)?(\d+)", query, re.IGNORECASE)
|
||
if not m:
|
||
raise ValueError(f"Не удалось определить номер RFC из строки: '{query}'")
|
||
num = int(m.group(1))
|
||
if num <= 0 or num > 100000:
|
||
raise ValueError(f"Недопустимый номер RFC: {num}")
|
||
return num
|
||
|
||
|
||
def parse_rfc_text(
|
||
raw_text: str,
|
||
rfc_number: int,
|
||
title: str = "",
|
||
author: str = ""
|
||
) -> Tuple[str, str, List[Dict[str, str]], int]:
|
||
"""
|
||
Cleans raw RFC text, strips pagination artifacts and form feeds,
|
||
and splits the specification into structured readable chapters.
|
||
Returns: (clean_title, clean_author, chapters, total_words)
|
||
"""
|
||
raw_text = raw_text.replace("\r\n", "\n").replace("\r", "\n").lstrip("\ufeff")
|
||
|
||
# 1. Clean pagination artifacts (running footers, form-feed \x0c, running headers)
|
||
clean = re.sub(
|
||
r"\n[^\n]*\[Page [0-9ivxlc]+\]\s*[\n\x0c]+\s*(?:RFC[^\n]*\n)?",
|
||
"\n\n",
|
||
raw_text,
|
||
flags=re.IGNORECASE
|
||
)
|
||
clean = clean.replace("\x0c", "\n\n")
|
||
|
||
lines = clean.splitlines()
|
||
|
||
# 2. If title wasn't provided, extract from the document header
|
||
if not title:
|
||
for l in lines[0:65]:
|
||
s = l.strip()
|
||
if not s:
|
||
continue
|
||
if re.match(
|
||
r"^(Internet Engineering Task Force|Request for Comments|STD:|Updates:|Obsoletes:|Category:|ISSN:|Network Working Group|BCP:|FYI:|RFC\s*\d+)",
|
||
s,
|
||
re.IGNORECASE
|
||
):
|
||
continue
|
||
if re.match(r"^Table of Contents\b", s, re.IGNORECASE):
|
||
break
|
||
if re.match(r"^Abstract\b", s, re.IGNORECASE):
|
||
break
|
||
if 4 <= len(s) <= 80 and not s.endswith((":", ";", ".")):
|
||
title = s
|
||
break
|
||
|
||
clean_title = f"RFC {rfc_number}: {title}" if title and not title.lower().startswith("rfc") else (title or f"RFC {rfc_number}")
|
||
clean_author = author or "IETF"
|
||
|
||
# 3. Detect chapters
|
||
in_toc = False
|
||
past_toc = False
|
||
headings = []
|
||
bad_sentence_starters = {
|
||
"if", "the", "when", "where", "this", "that", "these", "those",
|
||
"allow", "rather", "a", "an", "in", "for", "on", "to", "there",
|
||
"clients", "optionally", "both", "all", "each", "such", "see", "note"
|
||
}
|
||
|
||
for i, line in enumerate(lines):
|
||
trimmed = line.strip()
|
||
if not trimmed:
|
||
continue
|
||
|
||
if re.search(r"\.{4,}", line):
|
||
in_toc = True
|
||
continue
|
||
|
||
if not past_toc:
|
||
if re.match(r"^Table of Contents\b", trimmed, re.IGNORECASE):
|
||
in_toc = True
|
||
continue
|
||
if in_toc:
|
||
if (line.startswith("1.") or line.startswith("1 ") or re.match(r"^\s{10,}1\.\s+[A-Z]", line)) and not re.search(r"\.{3,}", line) and not re.search(r"\s+\d+$", trimmed):
|
||
in_toc = False
|
||
past_toc = True
|
||
else:
|
||
continue
|
||
|
||
prev_blank = (i == 0) or (not lines[i - 1].strip())
|
||
if not prev_blank:
|
||
continue
|
||
|
||
m = re.match(
|
||
r"^(?:([0-9]{1,2}\.?)\s+([A-Za-z][^\n]{1,65})|(Appendix\s+[A-Z]\.?\s*[^\n]{0,65})|(Authors?'? Addresses\b.*)|(Full Copyright Statement\b.*))$",
|
||
trimmed,
|
||
re.IGNORECASE
|
||
)
|
||
if not m and re.match(r"^[0-9]{1,2}\.\s+[A-Z\s]{3,50}$", trimmed):
|
||
m = True
|
||
|
||
if m:
|
||
if trimmed.endswith((".", ",", ":", ";", "-")):
|
||
continue
|
||
if len(trimmed) > 70:
|
||
continue
|
||
words = re.findall(r"\b[A-Za-z]+\b", trimmed)
|
||
if len(words) >= 2:
|
||
sec_word = words[0].lower() if not words[0].isdigit() else (words[1].lower() if len(words) > 1 else "")
|
||
if sec_word in bad_sentence_starters:
|
||
continue
|
||
headings.append((i, trimmed))
|
||
|
||
# Deduplicate headings that are right next to each other
|
||
deduped_headings = []
|
||
for h in headings:
|
||
if not deduped_headings or (h[0] - deduped_headings[-1][0] >= 3):
|
||
deduped_headings.append(h)
|
||
|
||
chapters: List[Dict[str, str]] = []
|
||
if deduped_headings:
|
||
first_line_idx = deduped_headings[0][0]
|
||
intro_content = "\n".join(lines[:first_line_idx]).strip()
|
||
if count_words(intro_content) >= 30:
|
||
chapters.append({
|
||
"title": "Overview & Abstract",
|
||
"content": intro_content
|
||
})
|
||
|
||
for idx, (line_idx, h_title) in enumerate(deduped_headings):
|
||
next_line_idx = deduped_headings[idx + 1][0] if idx + 1 < len(deduped_headings) else len(lines)
|
||
body = "\n".join(lines[line_idx:next_line_idx]).strip()
|
||
if body:
|
||
chapters.append({
|
||
"title": h_title,
|
||
"content": body
|
||
})
|
||
|
||
if not chapters:
|
||
paras = [p.strip() for p in re.split(r"\n{2,}", clean) if p.strip()]
|
||
chunks = []
|
||
curr = []
|
||
curr_w = 0
|
||
for p in paras:
|
||
w = count_words(p)
|
||
if curr_w + w > 1500 and curr:
|
||
chunks.append("\n\n".join(curr))
|
||
curr = [p]
|
||
curr_w = w
|
||
else:
|
||
curr.append(p)
|
||
curr_w += w
|
||
if curr:
|
||
chunks.append("\n\n".join(curr))
|
||
|
||
for idx, chunk in enumerate(chunks, 1):
|
||
chapters.append({
|
||
"title": f"Section {idx}",
|
||
"content": chunk
|
||
})
|
||
|
||
total_words = sum(count_words(ch["content"]) for ch in chapters)
|
||
return clean_title, clean_author, chapters, total_words
|
||
|
||
|
||
def fetch_and_parse_rfc(query: str) -> Tuple[str, str, List[Dict[str, str]], int, int]:
|
||
"""
|
||
Fetches an RFC by number or URL from RFC Editor & IETF Datatracker,
|
||
cleans formatting, parses chapters, and returns:
|
||
(clean_title, author, chapters, rfc_number, total_words)
|
||
"""
|
||
rfc_num = extract_rfc_number(query)
|
||
|
||
# 1. Fetch metadata from IETF Datatracker (title, authors)
|
||
dt_title = ""
|
||
dt_authors = []
|
||
try:
|
||
dt_url = f"https://datatracker.ietf.org/doc/rfc{rfc_num}/doc.json"
|
||
req = urllib.request.Request(dt_url, headers={"User-Agent": "study-qpid-pro/1.0 (educational-reader)"})
|
||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||
data = json.loads(resp.read().decode("utf-8", errors="replace"))
|
||
dt_title = (data.get("title") or "").strip()
|
||
for a in data.get("authors", []):
|
||
name = a.get("name") or a.get("person", {}).get("plain_name")
|
||
if name and name not in dt_authors:
|
||
dt_authors.append(name.strip())
|
||
except Exception:
|
||
pass
|
||
|
||
# 2. Fetch raw text from RFC Editor (primary) or Datatracker (fallback)
|
||
raw_text = None
|
||
urls_to_try = [
|
||
f"https://www.rfc-editor.org/rfc/rfc{rfc_num}.txt",
|
||
f"https://datatracker.ietf.org/doc/rfc{rfc_num}.txt",
|
||
]
|
||
fetch_err = None
|
||
for u in urls_to_try:
|
||
try:
|
||
req = urllib.request.Request(u, headers={"User-Agent": "study-qpid-pro/1.0 (educational-reader)"})
|
||
with urllib.request.urlopen(req, timeout=8) as resp:
|
||
raw_bytes = resp.read()
|
||
raw_text = raw_bytes.decode("utf-8", errors="replace")
|
||
break
|
||
except urllib.error.HTTPError as he:
|
||
fetch_err = he
|
||
if he.code == 404:
|
||
continue
|
||
except Exception as e:
|
||
fetch_err = e
|
||
continue
|
||
|
||
if not raw_text:
|
||
if isinstance(fetch_err, urllib.error.HTTPError) and fetch_err.code == 404:
|
||
raise ValueError(f"RFC {rfc_num} не найден в базе RFC Editor (404 Not Found)")
|
||
err_msg = str(fetch_err) if fetch_err else "документ недоступен"
|
||
raise ValueError(f"Не удалось загрузить RFC {rfc_num}: {err_msg}")
|
||
|
||
author = ("IETF (" + ", ".join(dt_authors) + ")") if dt_authors else "IETF"
|
||
clean_title, clean_author, chapters, total_words = parse_rfc_text(
|
||
raw_text, rfc_number=rfc_num, title=dt_title, author=author
|
||
)
|
||
|
||
return clean_title, clean_author, chapters, rfc_num, total_words
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Wikipedia Autonomous Import & Reader Module
|
||
# ---------------------------------------------------------------------------
|
||
|
||
WIKIPEDIA_USER_AGENT = "StudyReaderApp/1.0 (https://study.qpid.pro; admin@study.qpid.pro) edu-reader"
|
||
|
||
WIKIPEDIA_TOPIC_POOLS: Dict[str, Dict[str, List[str]]] = {
|
||
"science": {
|
||
"simple": [
|
||
"Solar_System", "Sun", "Moon", "Mars", "Black_hole", "Milky_Way", "Speed_of_light",
|
||
"Gravity", "Atom", "Dinosaur", "Evolution", "Photosynthesis", "Asteroid", "Comet",
|
||
"Volcano", "Plate_tectonics", "Hubble_Space_Telescope", "James_Webb_Space_Telescope",
|
||
"Planet", "Universe", "Earthquake", "Atmosphere"
|
||
],
|
||
"en": [
|
||
"Solar_System", "Sun", "Moon", "Mars", "Black_hole", "Milky_Way", "General_relativity",
|
||
"Quantum_mechanics", "Speed_of_light", "Plate_tectonics", "Supernova", "Exoplanet",
|
||
"Dark_matter", "Hubble_Space_Telescope", "James_Webb_Space_Telescope", "Pendulum",
|
||
"Thermodynamics", "Stellar_evolution", "Gravitational_wave"
|
||
]
|
||
},
|
||
"tech": {
|
||
"simple": [
|
||
"Computer", "Internet", "World_Wide_Web", "Artificial_intelligence", "Smartphone",
|
||
"Robot", "Video_game", "Software", "Linux", "Algorithm", "Satellite", "Electricity",
|
||
"Machine_learning", "Computer_network", "Fiber_optic", "Microprocessor"
|
||
],
|
||
"en": [
|
||
"Artificial_intelligence", "Computer", "Internet", "World_Wide_Web", "Quantum_computing",
|
||
"Machine_learning", "Operating_system", "Cryptography", "Blockchain", "Transistor",
|
||
"Linux", "Neural_network", "Optical_fiber", "Semiconductor"
|
||
]
|
||
},
|
||
"history": {
|
||
"simple": [
|
||
"Ancient_Egypt", "Roman_Empire", "Ancient_Greece", "Middle_Ages", "Industrial_Revolution",
|
||
"Silk_Road", "Renaissance", "Aztec", "World_War_I", "French_Revolution", "Great_Wall_of_China",
|
||
"Mesopotamia", "Byzantine_Empire", "Inca_Empire", "Voyages_of_Christopher_Columbus"
|
||
],
|
||
"en": [
|
||
"Ancient_Egypt", "Roman_Empire", "Renaissance", "Industrial_Revolution", "Silk_Road",
|
||
"Byzantine_Empire", "Age_of_Discovery", "Classical_Athens", "Mongol_Empire", "French_Revolution",
|
||
"Machu_Picchu", "Hanseatic_League", "Printing_press", "Maya_civilization"
|
||
]
|
||
},
|
||
"nature": {
|
||
"simple": [
|
||
"Blue_whale", "Tiger", "African_elephant", "Honey_bee", "Amazon_rainforest", "Coral_reef",
|
||
"Penguin", "Dolphin", "Wolf", "Sahara", "Giant_panda", "Cheetah", "Kangaroo", "Giraffe",
|
||
"Ocean", "Forest", "Rainforest", "Migration"
|
||
],
|
||
"en": [
|
||
"Blue_whale", "Tiger", "African_elephant", "Amazon_rainforest", "Great_Barrier_Reef",
|
||
"Deep_sea", "Polar_bear", "Galapagos_Islands", "Bioluminescence", "Coast_redwood",
|
||
"Honey_bee", "Monarch_butterfly", "Humpback_whale", "Ecosystem"
|
||
]
|
||
},
|
||
"psychology": {
|
||
"simple": [
|
||
"Memory", "Emotion", "Consciousness", "Sleep", "Brain", "Intelligence", "Dream",
|
||
"Perception", "Learning", "Happiness", "Stress_(biology)", "Mind", "Behavior"
|
||
],
|
||
"en": [
|
||
"Human_memory", "Emotion", "Consciousness", "Sleep", "Cognitive_bias", "Neuroplasticity",
|
||
"Empathy", "Perception", "Motivation", "Decision-making", "Working_memory",
|
||
"Attention", "Lateralization_of_brain_function"
|
||
]
|
||
}
|
||
}
|
||
|
||
|
||
def extract_wikipedia_title_and_lang(query: str, default_lang: str = "simple") -> Tuple[str, str]:
|
||
"""
|
||
Parses a Wikipedia URL or plain title into (article_title, lang).
|
||
Examples:
|
||
https://simple.wikipedia.org/wiki/Solar_System -> ('Solar_System', 'simple')
|
||
https://en.wikipedia.org/wiki/Quantum_mechanics -> ('Quantum_mechanics', 'en')
|
||
'Quantum mechanics' -> ('Quantum_mechanics', default_lang)
|
||
"""
|
||
raw = (query or "").strip()
|
||
if not raw:
|
||
return "", default_lang
|
||
|
||
# Check URL format
|
||
m_url = re.search(r"https?://([a-z]+)\.(?:m\.)?wikipedia\.org/wiki/([^?#]+)", raw, re.IGNORECASE)
|
||
if m_url:
|
||
lang = m_url.group(1).lower()
|
||
title_part = urllib.parse.unquote(m_url.group(2)).strip()
|
||
# Convert spaces
|
||
clean_title = title_part.replace(" ", "_")
|
||
return clean_title, lang
|
||
|
||
# Plain title or search query
|
||
clean_title = raw.replace(" ", "_")
|
||
return clean_title, default_lang
|
||
|
||
|
||
def clean_wiki_html(raw_html: str, article_title: str = "") -> Tuple[List[Dict[str, str]], str, int]:
|
||
"""
|
||
Parses MediaWiki Parsoid HTML into clean, structured chapters for the interactive reader.
|
||
Extracts cover image/thumbnail, preserves figures, animated GIFs/SVGs, and clean data tables,
|
||
and strips wiki clutter ([1], [citation needed], infobox rows, navboxes).
|
||
"""
|
||
cover_image = ""
|
||
|
||
# 1. Lead image: check infobox or first figure
|
||
lead_img_m = re.search(
|
||
r'<table\b[^>]*class=["\x27][^"\x27]*infobox[^"\x27]*["\x27][^>]*>.*?<img\b([^>]*)>',
|
||
raw_html,
|
||
re.DOTALL | re.IGNORECASE
|
||
)
|
||
if not lead_img_m:
|
||
lead_img_m = re.search(r'<figure\b[^>]*>.*?<img\b([^>]*)>', raw_html, re.DOTALL | re.IGNORECASE)
|
||
|
||
if lead_img_m:
|
||
src_m = re.search(r'src=["\x27]([^\s"\x27]+)["\x27]', lead_img_m.group(1), re.IGNORECASE)
|
||
if src_m:
|
||
src = src_m.group(1)
|
||
if src.startswith("//"):
|
||
src = "https:" + src
|
||
elif not src.startswith("http"):
|
||
src = "https://" + src.lstrip("/")
|
||
cover_image = src
|
||
|
||
# 2. Extract body
|
||
body_m = re.search(r"<body[^>]*>(.*?)</body>", raw_html, re.DOTALL | re.IGNORECASE)
|
||
body = body_m.group(1) if body_m else raw_html
|
||
|
||
# Strip style, script, noscript tags
|
||
body = re.sub(r'<style\b[^>]*>.*?</style>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
body = re.sub(r'<script\b[^>]*>.*?</script>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
body = re.sub(r'<noscript\b[^>]*>.*?</noscript>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
# 3. Strip unwanted wiki elements
|
||
# Remove references: <sup class="reference">...</sup> and bracketed reference markers
|
||
body = re.sub(r'<sup\b[^>]*class=["\x27][^"\x27]*reference[^"\x27]*["\x27][^>]*>.*?</sup>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
body = re.sub(r'<sup\b[^>]*>\s*\[\s*[\d\w\s,-]+\s*\]\s*</sup>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
body = re.sub(r'<span\b[^>]*class=["\x27][^"\x27]*mw-editsection[^"\x27]*["\x27][^>]*>.*?</span>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
body = re.sub(r'<div\b[^>]*class=["\x27][^"\x27]*(?:navbox|ambox|metadata|hatnote|sistersitebox|portal)[^"\x27]*["\x27][^>]*>.*?</div>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
body = re.sub(r'<table\b[^>]*class=["\x27][^"\x27]*(?:navbox|ambox|metadata|nowraplinks)[^"\x27]*["\x27][^>]*>.*?</table>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
# Remove infobox entirely from text flow (lead image was already captured)
|
||
body = re.sub(r'<table\b[^>]*class=["\x27][^"\x27]*infobox[^"\x27]*["\x27][^>]*>.*?</table>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
body = re.sub(r'<div\b[^>]*class=["\x27][^"\x27]*infobox[^"\x27]*["\x27][^>]*>.*?</div>', '', body, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
# 4. Standardize figures and images (preserving animated GIFs/SVGs/PNGs)
|
||
def clean_figure(m: re.Match) -> str:
|
||
f_html = m.group(0)
|
||
img_m = re.search(r'<img\b([^>]*)>', f_html, re.IGNORECASE)
|
||
if not img_m:
|
||
return ""
|
||
attrs = img_m.group(1)
|
||
src_m = re.search(r'src=["\x27]([^\s"\x27]+)["\x27]', attrs, re.IGNORECASE)
|
||
if not src_m:
|
||
return ""
|
||
src = src_m.group(1)
|
||
if src.startswith("//"):
|
||
src = "https:" + src
|
||
elif not src.startswith("http"):
|
||
src = "https://" + src.lstrip("/")
|
||
|
||
alt_m = re.search(r'alt=["\x27]([^\"]*)["\x27]', attrs, re.IGNORECASE)
|
||
alt = alt_m.group(1) if alt_m else ""
|
||
|
||
cap_m = re.search(r'<figcaption\b[^>]*>(.*?)</figcaption>', f_html, re.DOTALL | re.IGNORECASE)
|
||
cap = ""
|
||
if cap_m:
|
||
cap_clean = re.sub(r'<[^>]+>', ' ', cap_m.group(1))
|
||
cap = " ".join(cap_clean.split()).strip()
|
||
|
||
return f'\n\n<figure class="reader-figure"><img src="{src}" alt="{html.escape(alt)}" loading="lazy" /><figcaption>{html.escape(cap)}</figcaption></figure>\n\n'
|
||
|
||
body = re.sub(r'<figure\b[^>]*>.*?</figure>', clean_figure, body, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
# 5. Clean data tables (tables that are not navboxes/infoboxes)
|
||
def clean_data_table(m: re.Match) -> str:
|
||
tbl_html = m.group(0)
|
||
if re.search(r'class=["\x27][^"\x27]*(?:navbox|ambox|metadata|nowraplinks|infobox)[^"\x27]*["\x27]', tbl_html, re.IGNORECASE):
|
||
return ""
|
||
cleaned_tbl = re.sub(
|
||
r'\s+(?:style|bgcolor|align|valign|width|height|border|cellspacing|cellpadding|id)=["\x27][^"\x27]*["\x27]',
|
||
'',
|
||
tbl_html,
|
||
flags=re.IGNORECASE
|
||
)
|
||
cleaned_tbl = re.sub(r'<table[^>]*>', '<div class="reader-table-wrap"><table class="reader-table">', cleaned_tbl, count=1, flags=re.IGNORECASE)
|
||
cleaned_tbl = re.sub(r'</table>', '</table></div>', cleaned_tbl, count=1, flags=re.IGNORECASE)
|
||
return f'\n\n{cleaned_tbl}\n\n'
|
||
|
||
body = re.sub(r'<table\b[^>]*>.*?</table>', clean_data_table, body, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
# 6. Split into sections by <h2>
|
||
ignore_sections = {
|
||
"references", "reference", "see also", "external links", "further reading",
|
||
"notes", "footnotes", "sources", "bibliography", "other websites",
|
||
"related pages", "citations", "works cited", "navigation menu"
|
||
}
|
||
|
||
parts = re.split(r'(<h2\b[^>]*>.*?</h2>)', body, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
raw_chapters: List[Tuple[str, str]] = []
|
||
lead_content = parts[0].strip()
|
||
if lead_content:
|
||
raw_chapters.append(("Overview", lead_content))
|
||
|
||
idx = 1
|
||
while idx < len(parts):
|
||
h2_tag = parts[idx]
|
||
h_title = re.sub(r'<[^>]+>', '', h2_tag).strip()
|
||
h_title_clean = " ".join(h_title.split())
|
||
sec_content = parts[idx + 1].strip() if idx + 1 < len(parts) else ""
|
||
idx += 2
|
||
|
||
if h_title_clean.lower() in ignore_sections:
|
||
continue
|
||
|
||
raw_chapters.append((h_title_clean, sec_content))
|
||
|
||
# 7. Convert HTML within each chapter into clean reader content
|
||
chapters: List[Dict[str, str]] = []
|
||
|
||
def format_section_body(sec_html: str) -> str:
|
||
s = re.sub(r'<h3\b[^>]*>(.*?)</h3>', r'\n\n### \1\n\n', sec_html, flags=re.DOTALL | re.IGNORECASE)
|
||
s = re.sub(r'<h4\b[^>]*>(.*?)</h4>', r'\n\n#### \1\n\n', s, flags=re.DOTALL | re.IGNORECASE)
|
||
s = re.sub(r'<li\b[^>]*>(.*?)</li>', r'\n- \1', s, flags=re.DOTALL | re.IGNORECASE)
|
||
s = re.sub(r'</?[uo]l[^>]*>', '\n', s, flags=re.IGNORECASE)
|
||
s = re.sub(r'<p\b[^>]*>(.*?)</p>', r'\n\n\1\n\n', s, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
# Protect <figure...> and <div class="reader-table-wrap">
|
||
blocks_protected = []
|
||
def stash_block(match: re.Match) -> str:
|
||
token = f"§§BLOCK_{len(blocks_protected)}§§"
|
||
blocks_protected.append(match.group(0))
|
||
return f"\n\n{token}\n\n"
|
||
|
||
s = re.sub(r'<figure\b[^>]*>.*?</figure>', stash_block, s, flags=re.DOTALL | re.IGNORECASE)
|
||
s = re.sub(r'<div class="reader-table-wrap">.*?</div>', stash_block, s, flags=re.DOTALL | re.IGNORECASE)
|
||
|
||
s = re.sub(r'<[^>]+>', ' ', s)
|
||
s = html.unescape(s)
|
||
|
||
lines = []
|
||
for line in s.splitlines():
|
||
trimmed = " ".join(line.split())
|
||
if trimmed:
|
||
lines.append(trimmed)
|
||
elif lines and lines[-1] != "":
|
||
lines.append("")
|
||
|
||
res_text = "\n".join(lines).strip()
|
||
|
||
for b_idx, block_html in enumerate(blocks_protected):
|
||
res_text = res_text.replace(f"§§BLOCK_{b_idx}§§", f"\n\n{block_html}\n\n")
|
||
|
||
res_text = re.sub(r'\n{3,}', '\n\n', res_text).strip()
|
||
return res_text
|
||
|
||
for title, content_html in raw_chapters:
|
||
cleaned_body = format_section_body(content_html)
|
||
w_count = count_words(cleaned_body)
|
||
if w_count >= 15 or "<figure" in cleaned_body or "<table" in cleaned_body:
|
||
chapters.append({
|
||
"title": title,
|
||
"content": cleaned_body
|
||
})
|
||
|
||
if not chapters and raw_chapters:
|
||
for title, content_html in raw_chapters:
|
||
cleaned_body = format_section_body(content_html)
|
||
if cleaned_body.strip():
|
||
chapters.append({"title": title, "content": cleaned_body.strip()})
|
||
|
||
total_words = sum(count_words(c["content"]) for c in chapters)
|
||
return chapters, cover_image, total_words
|
||
|
||
|
||
def fetch_and_parse_wikipedia(
|
||
title_or_url: Optional[str] = None,
|
||
lang: str = "simple",
|
||
time_budget: str = "10-15",
|
||
topic: str = "random",
|
||
) -> Tuple[str, str, List[Dict[str, str]], str, int, str]:
|
||
"""
|
||
Fetches a Wikipedia article (by title/URL or by random topic & time budget),
|
||
parses into chapters with preserved media and tables, and returns:
|
||
(clean_title, author, chapters, cover_url, total_words, wiki_url)
|
||
"""
|
||
import random
|
||
from datetime import datetime
|
||
|
||
clean_lang = "en" if (lang or "").lower() in {"en", "standard", "standard english"} else "simple"
|
||
target_title = ""
|
||
|
||
# Case 1: Specific title or URL provided
|
||
if title_or_url and title_or_url.strip():
|
||
extracted_title, detected_lang = extract_wikipedia_title_and_lang(title_or_url, default_lang=clean_lang)
|
||
target_title = extracted_title
|
||
if detected_lang:
|
||
clean_lang = detected_lang
|
||
|
||
# Case 2: Pick random article by topic and time budget
|
||
if not target_title:
|
||
candidates = []
|
||
clean_topic = (topic or "random").lower().strip()
|
||
|
||
# Handle 'featured' article of the day
|
||
if clean_topic == "featured":
|
||
try:
|
||
now = datetime.utcnow()
|
||
feat_url = f"https://{clean_lang}.wikipedia.org/api/rest_v1/feed/featured/{now.year}/{now.month:02d}/{now.day:02d}"
|
||
req = urllib.request.Request(feat_url, headers={"User-Agent": WIKIPEDIA_USER_AGENT})
|
||
with urllib.request.urlopen(req, timeout=5) as r:
|
||
feed_data = json.loads(r.read().decode("utf-8"))
|
||
if "tfa" in feed_data and feed_data["tfa"].get("title"):
|
||
target_title = feed_data["tfa"]["title"]
|
||
elif "mostread" in feed_data and feed_data["mostread"].get("articles"):
|
||
art_list = [a["title"] for a in feed_data["mostread"]["articles"] if a.get("title")]
|
||
if art_list:
|
||
target_title = random.choice(art_list[:5])
|
||
except Exception:
|
||
pass
|
||
|
||
if not target_title:
|
||
if clean_topic in WIKIPEDIA_TOPIC_POOLS:
|
||
candidates = list(WIKIPEDIA_TOPIC_POOLS[clean_topic].get(clean_lang, []))
|
||
else:
|
||
for pool_dict in WIKIPEDIA_TOPIC_POOLS.values():
|
||
candidates.extend(pool_dict.get(clean_lang, []))
|
||
|
||
random.shuffle(candidates)
|
||
target_title = candidates[0] if candidates else "Solar_System"
|
||
|
||
# Fetch article summary to get clean title and thumbnail
|
||
clean_title = target_title.replace("_", " ")
|
||
cover_url = ""
|
||
wiki_url = f"https://{clean_lang}.wikipedia.org/wiki/{urllib.parse.quote(target_title)}"
|
||
|
||
try:
|
||
sum_url = f"https://{clean_lang}.wikipedia.org/api/rest_v1/page/summary/{urllib.parse.quote(target_title)}"
|
||
req = urllib.request.Request(sum_url, headers={"User-Agent": WIKIPEDIA_USER_AGENT})
|
||
with urllib.request.urlopen(req, timeout=8) as r:
|
||
sum_data = json.loads(r.read().decode("utf-8"))
|
||
if sum_data.get("title"):
|
||
clean_title = sum_data["title"]
|
||
if sum_data.get("thumbnail") and sum_data["thumbnail"].get("source"):
|
||
cover_url = sum_data["thumbnail"]["source"]
|
||
except urllib.error.HTTPError as he:
|
||
if he.code == 404:
|
||
raise ValueError(f"Статья «{clean_title}» не найдена в Википедии ({clean_lang}.wikipedia.org)")
|
||
except Exception:
|
||
pass
|
||
|
||
# Fetch full Parsoid HTML
|
||
raw_html = ""
|
||
try:
|
||
html_url = f"https://{clean_lang}.wikipedia.org/api/rest_v1/page/html/{urllib.parse.quote(target_title)}"
|
||
req = urllib.request.Request(html_url, headers={"User-Agent": WIKIPEDIA_USER_AGENT})
|
||
with urllib.request.urlopen(req, timeout=12) as r:
|
||
raw_html = r.read().decode("utf-8", errors="replace")
|
||
except urllib.error.HTTPError as he:
|
||
if he.code == 404:
|
||
raise ValueError(f"Статья «{clean_title}» не найдена в Википедии ({clean_lang}.wikipedia.org)")
|
||
raise ValueError(f"Ошибка загрузки статьи из Wikipedia: HTTP {he.code}")
|
||
except Exception as e:
|
||
raise ValueError(f"Не удалось соединиться с сервером Википедии: {e}")
|
||
|
||
chapters, html_cover, total_words = clean_wiki_html(raw_html, article_title=clean_title)
|
||
if not cover_url and html_cover:
|
||
cover_url = html_cover
|
||
|
||
if not chapters:
|
||
raise ValueError(f"В статье «{clean_title}» не найдено текстового содержимого для чтения")
|
||
|
||
edition_name = "Simple English" if clean_lang == "simple" else "Standard English"
|
||
author = f"Wikipedia ({edition_name})"
|
||
|
||
return clean_title, author, chapters, cover_url, total_words, wiki_url
|
||
|
||
|
||
|