#!/usr/bin/env python3
"""
Excel → HTML 自动转换器
监听 /watch 目录，xlsx/xls 文件上传后自动转成 HTML 写入 /output
"""
import os
import time
import html
import logging
from pathlib import Path
import openpyxl
from watchdog.observers.polling import PollingObserver # 卷挂载用 polling
from watchdog.events import FileSystemEventHandler
WATCH_DIR = Path("/watch")
OUTPUT_DIR = Path("/output")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [converter] %(levelname)s %(message)s",
)
log = logging.getLogger(__name__)
def xlsx_to_html(src: Path, dst: Path) -> None:
wb = openpyxl.load_workbook(src, read_only=True, data_only=True)
parts = []
parts.append(
""
""
)
parts.append(f"{html.escape(src.name)}
")
for sheet in wb.worksheets:
parts.append(f"{html.escape(sheet.title)}
")
parts.append("")
first_row = True
for row in sheet.iter_rows(values_only=True):
tag = "th" if first_row else "td"
cells = "".join(
f"<{tag}>{html.escape(str(c)) if c is not None else ''}"
for c in row
)
parts.append(f"{cells}
")
first_row = False
parts.append("
")
parts.append("")
dst.write_text("\n".join(parts), encoding="utf-8")
log.info("Converted: %s -> %s", src.name, dst.name)
def convert_file(path: Path) -> None:
if path.suffix.lower() not in (".xlsx", ".xls"):
return
try:
out = OUTPUT_DIR / (path.stem + ".html")
xlsx_to_html(path, out)
except Exception as e:
log.error("Failed to convert %s: %s", path, e)
class ExcelHandler(FileSystemEventHandler):
def on_created(self, event):
if not event.is_directory:
convert_file(Path(event.src_path))
def on_modified(self, event):
if not event.is_directory:
convert_file(Path(event.src_path))
if __name__ == "__main__":
log.info("Converter started. Watching: %s", WATCH_DIR)
# 启动时转换已有文件
for f in WATCH_DIR.rglob("*.xlsx"):
convert_file(f)
for f in WATCH_DIR.rglob("*.xls"):
convert_file(f)
observer = PollingObserver(timeout=2)
observer.schedule(ExcelHandler(), str(WATCH_DIR), recursive=True)
observer.start()
try:
while True:
time.sleep(5)
except KeyboardInterrupt:
observer.stop()
observer.join()