159 lines
5.2 KiB
Python
159 lines
5.2 KiB
Python
# coding=utf-8
|
||
"""本地对象存储（静态文件目录）。
|
||
|
||
云机默认落盘：/data/tpweb
|
||
对外访问：https://cpsb.ruimeida.cc/tpweb/...
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
import os
|
||
import re
|
||
import time
|
||
from pathlib import Path
|
||
|
||
_FOLDER = os.path.dirname(os.path.abspath(__file__))
|
||
|
||
# 云微服务：上传走 :9100；落盘 /data/tpweb；对外 https://cpsb.ruimeida.cc/tpweb
|
||
CLOUD_MEDIA_HOST = "116.147.40.47"
|
||
CLOUD_MEDIA_PORT = 9100
|
||
# 部署目录参考：/home/cpsb/project/dish_server
|
||
# 磁盘目录（与 Nginx /tpweb 根一致）；本机 Windows 开发回落到项目内目录
|
||
_DEFAULT_MEDIA_ROOT = (
|
||
os.environ.get("DISH_MEDIA_ROOT")
|
||
or ("/data/tpweb" if os.name != "nt" else os.path.join(_FOLDER, "data", "tpweb"))
|
||
)
|
||
# 浏览器访问前缀（HTTPS，避免管理系统 Mixed Content）
|
||
CLOUD_MEDIA_PUBLIC_BASE = "https://cpsb.ruimeida.cc/tpweb"
|
||
CLOUD_MEDIA_UPLOAD_URL = f"http://{CLOUD_MEDIA_HOST}:{CLOUD_MEDIA_PORT}/api/dish/media/upload"
|
||
|
||
MEDIA_ROOT = _DEFAULT_MEDIA_ROOT
|
||
MEDIA_PUBLIC_BASE = os.environ.get("DISH_MEDIA_PUBLIC_BASE", CLOUD_MEDIA_PUBLIC_BASE).rstrip("/")
|
||
|
||
_SAFE = re.compile(r"[^A-Za-z0-9._-]+")
|
||
# template：学菜模板图，落盘 template/{菜名}/xxx.jpg
|
||
ALLOWED_TYPES = ("source", "labeled", "crop", "template")
|
||
ALLOWED_EXT = {".jpg", ".jpeg", ".png", ".bmp", ".webp"}
|
||
|
||
|
||
def _safe_token(value, fallback="x"):
|
||
text = _SAFE.sub("_", str(value or "").strip())[:80]
|
||
return text or fallback
|
||
|
||
|
||
def _safe_dish_dirname(value, fallback="unknown"):
|
||
"""菜名目录：保留中文，去掉路径非法字符。"""
|
||
text = str(value or "").strip()
|
||
for ch in '<>:"/\\|?*\x00':
|
||
text = text.replace(ch, "_")
|
||
text = text.strip(" .")[:80]
|
||
return text or fallback
|
||
|
||
|
||
def public_url(rel_path):
|
||
rel = str(rel_path or "").replace("\\", "/").lstrip("/")
|
||
return f"{MEDIA_PUBLIC_BASE}/{rel}"
|
||
|
||
|
||
def local_path(rel_path):
|
||
rel = str(rel_path or "").replace("\\", "/").lstrip("/")
|
||
return os.path.normpath(os.path.join(MEDIA_ROOT, *rel.split("/")))
|
||
|
||
|
||
def is_http_url(value):
|
||
text = (value or "").strip().lower()
|
||
return text.startswith("http://") or text.startswith("https://")
|
||
|
||
|
||
class MediaStore:
|
||
"""按 基地 / 请求号 归档图片。"""
|
||
|
||
def __init__(self, root=None, public_base=None):
|
||
self.root = os.path.abspath(root or MEDIA_ROOT)
|
||
self.public_base = (public_base or MEDIA_PUBLIC_BASE).rstrip("/")
|
||
os.makedirs(self.root, exist_ok=True)
|
||
|
||
def save_bytes(
|
||
self,
|
||
data,
|
||
filename,
|
||
media_type="source",
|
||
base_id="BASE",
|
||
request_id=None,
|
||
dish_name=None,
|
||
):
|
||
if not data:
|
||
raise ValueError("empty file")
|
||
media_type = (media_type or "source").strip().lower()
|
||
if media_type not in ALLOWED_TYPES:
|
||
media_type = "source"
|
||
ext = Path(filename or "").suffix.lower() or ".jpg"
|
||
if ext not in ALLOWED_EXT:
|
||
ext = ".jpg"
|
||
|
||
if media_type == "template":
|
||
# 学菜模板：/data/tpweb/template/{菜名}/文件名
|
||
dish = _safe_dish_dirname(dish_name or request_id or "unknown")
|
||
orig = Path(filename or f"{dish}{ext}").name
|
||
for ch in '<>:"/\\|?*\x00':
|
||
orig = orig.replace(ch, "_")
|
||
if not Path(orig).suffix:
|
||
orig = f"{orig}{ext}"
|
||
rel = f"template/{dish}/{orig}"
|
||
else:
|
||
req = _safe_token(request_id or f"REQ_{int(time.time() * 1000)}")
|
||
base = _safe_token(base_id or "BASE")
|
||
stem = _safe_token(Path(filename or media_type).stem, media_type)
|
||
rel = f"{base}/{req}/{media_type}_{stem}{ext}"
|
||
|
||
dest = local_path_under(self.root, rel)
|
||
os.makedirs(os.path.dirname(dest), exist_ok=True)
|
||
with open(dest, "wb") as fh:
|
||
fh.write(data)
|
||
rel_url = rel.replace("\\", "/")
|
||
return {
|
||
"rel_path": rel_url,
|
||
"path": dest,
|
||
"url": f"{self.public_base}/{rel_url}",
|
||
"type": media_type,
|
||
}
|
||
|
||
def save_file(
|
||
self,
|
||
src_path,
|
||
media_type="source",
|
||
base_id="BASE",
|
||
request_id=None,
|
||
dish_name=None,
|
||
):
|
||
if not src_path or not os.path.isfile(src_path):
|
||
return None
|
||
with open(src_path, "rb") as fh:
|
||
data = fh.read()
|
||
return self.save_bytes(
|
||
data,
|
||
filename=os.path.basename(src_path),
|
||
media_type=media_type,
|
||
base_id=base_id,
|
||
request_id=request_id,
|
||
dish_name=dish_name,
|
||
)
|
||
|
||
|
||
def local_path_under(root, rel_path):
|
||
dest = os.path.normpath(os.path.join(root, *str(rel_path).replace("\\", "/").split("/")))
|
||
root_abs = os.path.abspath(root)
|
||
dest_abs = os.path.abspath(dest)
|
||
if dest_abs != root_abs and not dest_abs.startswith(root_abs + os.sep):
|
||
raise ValueError("invalid media path")
|
||
return dest_abs
|
||
|
||
|
||
def copy_if_local(src_path, media_type, base_id, request_id, store=None):
|
||
if not src_path:
|
||
return src_path
|
||
if is_http_url(src_path):
|
||
return src_path
|
||
store = store or MediaStore()
|
||
saved = store.save_file(src_path, media_type=media_type, base_id=base_id, request_id=request_id)
|
||
return saved["url"] if saved else src_path
|