From 110b8a960cb1e69b26b064ee108d0d3de20d7a11 Mon Sep 17 00:00:00 2001 From: BXY0414 <1079823339@qq.com> Date: Wed, 22 Apr 2026 18:53:20 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=20=E4=B8=8A=E4=BC=A0?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + upload_inventory_page.py | 678 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 679 insertions(+) create mode 100644 upload_inventory_page.py diff --git a/.gitignore b/.gitignore index 74cf1af..de3a51b 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ .feishu_oauth_tokens.json inventory_sync/__pycache__ .DS_Store +batch_inventory_data/* diff --git a/upload_inventory_page.py b/upload_inventory_page.py new file mode 100644 index 0000000..232c063 --- /dev/null +++ b/upload_inventory_page.py @@ -0,0 +1,678 @@ +#!/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""" + + + + + 委外批次库存文件上传 + + + +
+
+
+
+
委委外批次库存上传台
+
每周约 2 次使用
+
+
+
+

投递委外批次库存文件

+

拖入委外仓、加工方或批次盘点文件，系统会按上传时间自动归档，便于后续同步、核对和追溯。

+
+
+ ARCHIVE PATH + {escape(str(UPLOAD_DIR))} +
+
+
+ +
+
+ {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())