Spaces:
Running
Running
File size: 3,904 Bytes
6de26b7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | """用户建议反馈 - HF 仓库读写"""
import json
import os
import tempfile
import uuid
from datetime import datetime
from config import FEEDBACK_REPO_PREFIX
from repo_util import download_repo_file, list_repo_files_safe, load_repo_json, upload_repo_file
def _parse_created(value):
if not value:
return datetime.min
try:
return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except ValueError:
try:
return datetime.strptime(str(value)[:19], "%Y-%m-%d %H:%M:%S")
except ValueError:
return datetime.min
def _normalize_feedback(meta):
meta = dict(meta or {})
meta.setdefault("feedback_id", "")
meta.setdefault("username", "匿名")
meta.setdefault("text", "")
meta.setdefault("is_public", False)
meta.setdefault("created", "")
meta.setdefault("images", [])
meta["image_count"] = len(meta.get("images") or [])
meta["public_label"] = "公开" if meta.get("is_public") else "仅管理员可见"
text = meta.get("text") or ""
meta["text_preview"] = text if len(text) <= 80 else text[:80] + "..."
return meta
def fetch_all_feedback():
"""获取全部用户反馈（本地管理端）"""
files = list_repo_files_safe()
prefix = FEEDBACK_REPO_PREFIX.rstrip("/") + "/"
json_files = [f for f in files if f.startswith(prefix) and f.endswith(".json")]
feedbacks = []
for json_path in json_files:
try:
meta = _normalize_feedback(load_repo_json(json_path))
feedbacks.append(meta)
except Exception:
continue
feedbacks.sort(key=lambda item: _parse_created(item.get("created")), reverse=True)
return feedbacks
def fetch_public_feedback():
"""获取允许公开展示的反馈（web 端）"""
return [item for item in fetch_all_feedback() if item.get("is_public")]
def submit_feedback(username, text, image_files=None, is_public=False):
"""提交用户反馈到 HF 仓库"""
text = (text or "").strip()
if not text:
raise ValueError("请填写反馈内容")
feedback_id = uuid.uuid4().hex
image_paths = []
image_files = image_files or []
for index, image_path in enumerate(image_files):
if not image_path or not os.path.exists(image_path):
continue
ext = os.path.splitext(image_path)[1].lower() or ".png"
remote_path = f"{FEEDBACK_REPO_PREFIX}/{feedback_id}_{index}{ext}"
upload_repo_file(image_path, remote_path)
image_paths.append(remote_path)
meta = {
"feedback_id": feedback_id,
"username": (username or "").strip() or "匿名",
"text": text,
"is_public": bool(is_public),
"created": datetime.now().isoformat(),
"images": image_paths,
}
with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".json", delete=False) as fp:
json.dump(meta, fp, ensure_ascii=False, indent=2)
tmp_json = fp.name
try:
upload_repo_file(tmp_json, f"{FEEDBACK_REPO_PREFIX}/{feedback_id}.json")
finally:
os.remove(tmp_json)
return meta
def get_feedback(feedback_id):
"""按 ID 获取单条反馈"""
json_path = f"{FEEDBACK_REPO_PREFIX}/{feedback_id}.json"
return _normalize_feedback(load_repo_json(json_path))
def ensure_feedback_image_cached(feedback_id, image_index, cache_dir):
"""下载反馈图片到本地缓存并返回路径"""
meta = get_feedback(feedback_id)
images = meta.get("images") or []
if image_index < 0 or image_index >= len(images):
raise FileNotFoundError("图片不存在")
remote_path = images[image_index]
filename = os.path.basename(remote_path)
local_path = os.path.join(cache_dir, f"{feedback_id}_{filename}")
if os.path.exists(local_path):
return local_path
return download_repo_file(remote_path, local_path)
|