委外批次库存文件上传

#!/usr/bin/env python3 """Local inventory file upload page. Run: python3 upload_inventory_page.py Then open: http://127.0.0.1:8080 """ from __future__ import annotations import cgi from datetime import datetime from html import escape from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import re import shutil from typing import Iterable from urllib.parse import parse_qs, urlparse HOST = "127.0.0.1" PORT = 8080 PROJECT_ROOT = Path(__file__).resolve().parent UPLOAD_DIR = PROJECT_ROOT / "batch_inventory_data" MAX_UPLOAD_BYTES = 200 * 1024 * 1024 def safe_filename(filename: str) -> str: name = Path(filename or "inventory_file").name.strip() name = re.sub(r"[^\w.\-\u4e00-\u9fff（）() ]+", "_", name) name = re.sub(r"\s+", "_", name) return name or "inventory_file" def unique_upload_path(filename: str) -> Path: UPLOAD_DIR.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") base_name = safe_filename(filename) candidate = UPLOAD_DIR / f"{timestamp}_{base_name}" counter = 1 while candidate.exists(): candidate = UPLOAD_DIR / f"{timestamp}_{counter}_{base_name}" counter += 1 return candidate def list_uploaded_files(limit: int = 20) -> list[Path]: UPLOAD_DIR.mkdir(parents=True, exist_ok=True) files = [path for path in UPLOAD_DIR.iterdir() if path.is_file() and not path.name.startswith(".")] return sorted(files, key=lambda item: item.stat().st_mtime, reverse=True)[:limit] def human_size(size: int) -> str: units = ["B", "KB", "MB", "GB"] value = float(size) for unit in units: if value < 1024 or unit == units[-1]: return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} {unit}" value /= 1024 return f"{size} B" def render_recent_files(files: Iterable[Path]) -> str: rows = [] for path in files: stat = path.stat() uploaded_at = datetime.fromtimestamp(stat.st_mtime).strftime("%Y-%m-%d %H:%M:%S") rows.append( "" f"{escape(path.name)}" f"{escape(human_size(stat.st_size))}" f"{escape(uploaded_at)}" "" ) if not rows: return '

还没有上传文件。上传成功后，最近文件会显示在这里。

' return ( '' "" f"{''.join(rows)}" "

文件名	大小	上传时间

" ) def render_page(message: str = "", is_error: bool = False) -> bytes: message_html = "" if message: message_class = "message error" if is_error else "message success" message_html = f'

{escape(message)}

' html = f""" 委外批次库存文件上传

{message_html}

⇣

DROP FILE HERE

拖入文件，立即上传

支持 Excel、CSV 等委外批次库存文件。松开鼠标后自动保存到 batch_inventory_data，无需点击选择文件。

等待拖入文件

最近上传的委外批次文件

{render_recent_files(list_uploaded_files())}

""" return html.encode("utf-8") class UploadHandler(BaseHTTPRequestHandler): server_version = "InventoryUpload/1.0" def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path == "/favicon.ico": self.send_response(204) self.end_headers() return if parsed.path != "/": self.send_error(404, "Not Found") return params = parse_qs(parsed.query) message = params.get("message", [""])[0] is_error = params.get("error", ["0"])[0] == "1" self.respond_html(render_page(message, is_error)) def do_POST(self) -> None: path = urlparse(self.path).path if path != "/upload": self.send_error(404, "Not Found") return content_length = int(self.headers.get("Content-Length", "0") or "0") if content_length <= 0: self.redirect("/", "没有收到上传文件。", is_error=True) return if content_length > MAX_UPLOAD_BYTES: self.redirect("/", f"文件太大，最大支持 {human_size(MAX_UPLOAD_BYTES)}。", is_error=True) return form = cgi.FieldStorage(fp=self.rfile, headers=self.headers, environ={ "REQUEST_METHOD": "POST", "CONTENT_TYPE": self.headers.get("Content-Type", ""), "CONTENT_LENGTH": str(content_length), }) file_item = form["inventory_file"] if "inventory_file" in form else None if file_item is None or not getattr(file_item, "filename", ""): self.redirect("/", "请选择需要上传的文件。", is_error=True) return target_path = unique_upload_path(file_item.filename) with target_path.open("wb") as output: shutil.copyfileobj(file_item.file, output) self.redirect("/", f"上传成功：{target_path.name}") def respond_html(self, body: bytes) -> None: self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def redirect(self, path: str, message: str, is_error: bool = False) -> None: from urllib.parse import urlencode query = urlencode({"message": message, "error": "1" if is_error else "0"}) self.send_response(303) self.send_header("Location", f"{path}?{query}") self.end_headers() def log_message(self, format: str, *args: object) -> None: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {self.address_string()} - {format % args}") def main() -> int: UPLOAD_DIR.mkdir(parents=True, exist_ok=True) server = ThreadingHTTPServer((HOST, PORT), UploadHandler) print(f"上传页面已启动: http://{HOST}:{PORT}") print(f"文件保存目录: {UPLOAD_DIR}") try: server.serve_forever() except KeyboardInterrupt: print("\n上传页面已停止。") finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())

投递委外批次库存文件

拖入文件，立即上传

最近上传的委外批次文件