"""
Export helpers extracted from the legacy web router.
"""
from __future__ import annotations
import logging
import os
import re
import tempfile
import urllib.parse
import urllib.request
import zipfile
from pathlib import Path
from typing import Any, Dict, List, Optional
from fastapi import Request
from pydantic import BaseModel
from ...core.config import ai_config
from ...services.pyppeteer_pdf_converter import get_pdf_converter
from ...utils.thread_pool import run_blocking_io
from .support import logger
def _coerce_bool(value) -> bool:
if isinstance(value, bool):
return value
if value is None:
return False
return str(value).strip().lower() in {"1", "true", "yes", "on"}
async def _is_standard_pptx_export_enabled() -> bool:
"""Return whether Apryse-based standard PPTX export is enabled system-wide."""
feature_enabled = _coerce_bool(getattr(ai_config, "enable_apryse_pptx_export", False))
license_key = str(getattr(ai_config, "apryse_license_key", "") or "").strip()
try:
from ...services.db_config_service import get_db_config_service
config_service = get_db_config_service()
system_config = await config_service.get_all_config(user_id=None)
feature_enabled = _coerce_bool(system_config.get("enable_apryse_pptx_export", feature_enabled))
license_key = str(system_config.get("apryse_license_key") or license_key or "").strip()
except Exception as exc:
logger.warning("Failed to resolve Apryse PPTX export state from DB config: %s", exc)
return feature_enabled and bool(license_key)
def _strip_default_port(host: str, scheme: str) -> str:
"""Strip default port from a host string."""
host = (host or "").strip()
scheme = (scheme or "").strip().lower()
if not host:
return host
try:
parsed = urllib.parse.urlsplit(f"{scheme or 'http'}://{host}")
hostname = parsed.hostname or host
port = parsed.port
if port is None:
return host
if (scheme == "http" and port == 80) or (scheme == "https" and port == 443):
if ":" in hostname and not hostname.startswith("["):
return f"[{hostname}]"
return hostname
except Exception:
return host
return host
def _normalize_base_url_candidate(raw_url: Optional[str], default_scheme: str = "https") -> Optional[str]:
"""Normalize a base-URL candidate to scheme://host[:port]."""
value = (str(raw_url).strip() if raw_url is not None else "")
if not value:
return None
if value.startswith("//"):
value = f"{default_scheme}:{value}"
elif "://" not in value:
value = f"{default_scheme}://{value.lstrip('/')}"
parsed = urllib.parse.urlsplit(value)
scheme = (parsed.scheme or default_scheme or "https").strip().lower()
host = (parsed.netloc or "").strip()
if not host:
return None
if "," in host:
host = host.split(",", 1)[0].strip()
host = _strip_default_port(host, scheme)
if not host:
return None
return f"{scheme}://{host}"
def _is_loopback_base_url(base_url: Optional[str]) -> bool:
"""Return True when the URL points at a loopback/local host."""
if not base_url:
return True
try:
parsed = urllib.parse.urlsplit(base_url)
hostname = (parsed.hostname or "").strip().lower()
return hostname in {"localhost", "127.0.0.1", "0.0.0.0", "::1"}
except Exception:
return False
def _resolve_export_base_url(http_request: Optional[Request] = None) -> str:
"""Resolve the public base URL used by file-based export renderers."""
request_candidates: List[str] = []
config_candidates: List[str] = []
def add_candidate(target: List[str], raw_url: Optional[str], *, default_scheme: str = "https") -> None:
normalized = _normalize_base_url_candidate(raw_url, default_scheme=default_scheme)
if normalized and normalized not in target:
target.append(normalized)
try:
if http_request is not None:
headers = http_request.headers
request_scheme = (http_request.url.scheme or "https").strip().lower()
add_candidate(request_candidates, headers.get("origin"), default_scheme=request_scheme)
referer = headers.get("referer")
if referer:
try:
referer_parts = urllib.parse.urlsplit(referer)
add_candidate(request_candidates, f"{referer_parts.scheme}://{referer_parts.netloc}", default_scheme=request_scheme)
except Exception:
pass
forwarded_host = (headers.get("x-forwarded-host") or "").strip()
forwarded_proto = (headers.get("x-forwarded-proto") or request_scheme).strip().lower()
forwarded_port = (headers.get("x-forwarded-port") or "").strip()
if "," in forwarded_host:
forwarded_host = forwarded_host.split(",", 1)[0].strip()
if "," in forwarded_proto:
forwarded_proto = forwarded_proto.split(",", 1)[0].strip()
if forwarded_host and forwarded_port and ":" not in forwarded_host:
forwarded_host = f"{forwarded_host}:{forwarded_port}"
add_candidate(request_candidates, forwarded_host, default_scheme=forwarded_proto or request_scheme)
host = (headers.get("host") or http_request.url.netloc or "").strip()
if host:
add_candidate(request_candidates, host, default_scheme=request_scheme)
if getattr(http_request, "base_url", None):
add_candidate(request_candidates, str(http_request.base_url), default_scheme=request_scheme)
except Exception:
pass
try:
from ...services.url_service import get_current_base_url
add_candidate(config_candidates, get_current_base_url())
except Exception:
pass
for candidate in request_candidates:
if not _is_loopback_base_url(candidate):
return candidate
for candidate in config_candidates:
if not _is_loopback_base_url(candidate):
return candidate
if request_candidates:
return request_candidates[0]
raise ValueError("Unable to resolve export base URL from request headers or app configuration")
def _build_export_app_url(base_url: str, relative_path: str) -> str:
"""Build an absolute app URL using the resolved export base URL."""
normalized_path = "/" + (relative_path or "").lstrip("/")
return urllib.parse.urljoin(f"{base_url.rstrip('/')}/", normalized_path.lstrip("/"))
_APP_EXPORTABLE_PATH_PREFIXES = (
"/api/image/view/",
"/api/image/thumbnail/",
"/static/",
"/temp/",
)
def _is_app_exportable_path(path: str) -> bool:
normalized_path = "/" + str(path or "").lstrip("/")
return any(normalized_path.startswith(prefix) for prefix in _APP_EXPORTABLE_PATH_PREFIXES)
def _resolve_export_absolute_resource_url(raw_url: str, base_url: str) -> str:
"""将误写成宿主机 localhost 的应用资源地址改写为导出进程可达地址。"""
try:
parsed = urllib.parse.urlsplit(raw_url)
except Exception:
return raw_url
if (parsed.scheme or "").lower() not in {"http", "https"}:
return raw_url
hostname = (parsed.hostname or "").strip().lower()
if hostname not in {"localhost", "127.0.0.1", "0.0.0.0"}:
return raw_url
if not _is_app_exportable_path(parsed.path):
return raw_url
relative_path = parsed.path or "/"
if parsed.query:
relative_path = f"{relative_path}?{parsed.query}"
if parsed.fragment:
relative_path = f"{relative_path}#{parsed.fragment}"
return _build_export_app_url(base_url, relative_path)
def _file_url_to_path(raw_url: str) -> Optional[Path]:
"""Convert a file:// URL to a local filesystem path."""
try:
parsed = urllib.parse.urlsplit(raw_url)
if (parsed.scheme or "").lower() != "file":
return None
pathname = urllib.request.url2pathname(parsed.path or "")
if parsed.netloc and parsed.netloc.lower() != "localhost":
pathname = f"//{parsed.netloc}{pathname}"
if os.name == "nt" and pathname.startswith("/") and re.match(r"^/[A-Za-z]:[/\\\\]", pathname):
pathname = pathname[1:]
return Path(pathname).resolve(strict=False)
except Exception:
return None
def _resolve_export_file_resource_url(raw_url: str, base_url: str) -> str:
"""Rewrite app-owned local file URLs to public URLs usable from export renderers."""
local_path = _file_url_to_path(raw_url)
if local_path is None:
return raw_url
try:
static_root = (Path(__file__).resolve().parent.parent / "static").resolve()
if local_path.is_relative_to(static_root):
relative_path = local_path.relative_to(static_root).as_posix()
quoted = urllib.parse.quote(relative_path, safe="/")
return _build_export_app_url(base_url, f"/static/{quoted}")
except Exception:
pass
try:
from ...services.image.image_service import get_image_service
image_service = get_image_service()
cache_index = getattr(getattr(image_service, "cache_manager", None), "_cache_index", {}) or {}
for cache_key, cache_info in cache_index.items():
file_path = getattr(cache_info, "file_path", None)
if not file_path:
continue
try:
if Path(file_path).resolve(strict=False) == local_path:
quoted_key = urllib.parse.quote(str(cache_key), safe="")
return _build_export_app_url(base_url, f"/api/image/view/{quoted_key}")
except Exception:
continue
except Exception:
pass
return raw_url
def _resolve_export_resource_url(raw_url: str, base_url: str) -> str:
"""Convert export-time relative resource URLs into absolute URLs."""
if not isinstance(raw_url, str):
return raw_url
candidate = raw_url.strip()
if not candidate:
return raw_url
lowered = candidate.lower()
if lowered.startswith(("#", "data:", "blob:", "javascript:", "mailto:", "tel:", "about:")):
return raw_url
if lowered.startswith(("http://", "https://")):
return _resolve_export_absolute_resource_url(candidate, base_url)
if lowered.startswith("file://"):
return _resolve_export_file_resource_url(candidate, base_url)
if candidate.startswith("//"):
base_scheme = urllib.parse.urlparse(base_url).scheme or "http"
return f"{base_scheme}:{candidate}"
joined = urllib.parse.urljoin(f"{base_url.rstrip('/')}/", candidate)
return joined or candidate
def _rewrite_export_css_urls(css_text: str, base_url: str) -> str:
"""Rewrite url(...) references in CSS so file:// exports can still fetch assets."""
if not isinstance(css_text, str) or "url(" not in css_text.lower():
return css_text
def replace_match(match: re.Match) -> str:
prefix = match.group(1)
raw_value = (match.group(2) or "").strip()
suffix = match.group(3)
quote = ""
inner = raw_value
if len(raw_value) >= 2 and raw_value[0] == raw_value[-1] and raw_value[0] in ("'", '"'):
quote = raw_value[0]
inner = raw_value[1:-1]
absolute_url = _resolve_export_resource_url(inner, base_url)
return f"{prefix}{quote}{absolute_url}{quote}{suffix}"
return re.sub(r"(url\(\s*)([^)]+?)(\s*\))", replace_match, css_text, flags=re.IGNORECASE)
def _rewrite_export_srcset(srcset_value: str, base_url: str) -> str:
"""Rewrite each candidate in srcset to an absolute URL."""
if not isinstance(srcset_value, str) or not srcset_value.strip():
return srcset_value
rewritten_candidates: List[str] = []
for candidate in srcset_value.split(","):
item = candidate.strip()
if not item:
continue
parts = item.split()
if not parts:
continue
parts[0] = _resolve_export_resource_url(parts[0], base_url)
rewritten_candidates.append(" ".join(parts))
return ", ".join(rewritten_candidates)
def _html_uses_tailwind_utilities(html_content: str) -> bool:
"""Best-effort detection for Tailwind utility class usage."""
if not isinstance(html_content, str) or "class" not in html_content.lower():
return False
utility_pattern = re.compile(
r"^(?:"
r"container|sr-only|not-sr-only|block|inline|inline-block|inline-flex|flex|inline-grid|grid|hidden|contents|"
r"absolute|relative|fixed|sticky|static|"
r"(?:top|right|bottom|left|inset|z)-[\w./\[\]-]+|"
r"(?:m|mx|my|mt|mr|mb|ml|p|px|py|pt|pr|pb|pl|w|min-w|max-w|h|min-h|max-h|"
r"gap|space-x|space-y|basis|grow|shrink|order|col|row|"
r"text|font|leading|tracking|bg|from|via|to|border|rounded|shadow|opacity|"
r"items|justify|content|self|place|object|overflow|overscroll|whitespace|break|"
r"aspect|ring|fill|stroke|list|underline|line-clamp|animate|duration|delay|ease|"
r"scale|rotate|translate|skew)-[\w./:%\[\]-]+|"
r"(?:prose|antialiased|subpixel-antialiased|uppercase|lowercase|capitalize|truncate|underline|no-underline|italic|not-italic|"
r"pointer-events-none|pointer-events-auto|select-none|select-text|align-middle|align-top|align-bottom)"
r")$",
re.IGNORECASE,
)
for match in re.finditer(r'class\s*=\s*["\']([^"\']+)["\']', html_content, flags=re.IGNORECASE):
class_value = match.group(1) or ""
for token in re.split(r"\s+", class_value.strip()):
if token and utility_pattern.match(token):
return True
return False
def _strip_unused_tailwind_cdn(html_content: str) -> str:
"""Remove Tailwind Play CDN when the document does not appear to use Tailwind utilities."""
if not isinstance(html_content, str) or "cdn.tailwindcss.com" not in html_content.lower():
return html_content
if _html_uses_tailwind_utilities(html_content):
return html_content
cleaned = re.sub(
r'',
'',
html_content,
flags=re.IGNORECASE | re.DOTALL,
)
if cleaned != html_content:
cleaned = re.sub(
r'',
'',
cleaned,
flags=re.IGNORECASE | re.DOTALL,
)
return cleaned
def _prepare_html_for_file_based_export(html_content: str, base_url: str) -> str:
"""Normalize resource URLs before rendering HTML from a local temp file."""
if not isinstance(html_content, str) or not html_content.strip():
return html_content
prepared = _strip_unused_tailwind_cdn(html_content)
normalized_base_url = base_url.rstrip("/")
base_href = f"{normalized_base_url}/"
if re.search(r"
PPT演示文稿 - 共{len(slide_files)}页
错误信息: {str(e)}
请确保PPT已经生成完成后再尝试导出。
""" async def _generate_individual_slide_html(slide, slide_number: int, total_slides: int, topic: str) -> str: """Generate complete HTML document for individual slide preserving original styles""" slide_html = slide.get('html_content', '') slide_title = slide.get('title', f'第{slide_number}页') # Check if it's already a complete HTML document import re if slide_html.strip().lower().startswith('' enhanced_html = re.sub(body_pattern, navigation_html + '\n' + navigation_js + '\n', enhanced_html, flags=re.IGNORECASE) return enhanced_html async def _generate_pdf_slide_html(slide, slide_number: int, total_slides: int, topic: str) -> str: """Generate PDF-optimized HTML for individual slide without navigation elements""" slide_html = slide.get('html_content', '') slide_title = slide.get('title', f'第{slide_number}页') # Check if it's already a complete HTML document import re if slide_html.strip().lower().startswith('