dekun
88fe4bbe56
固定 5m/15m/1h/4h 四槽截图上传与详情四宫格展示；自动 K 线默认关闭且与手动上传互斥。 Co-authored-by: Cursor <cursoragent@cursor.com>
120 lines
3.8 KiB
Python
120 lines
3.8 KiB
Python
"""复盘记录：多周期截图上传、存储与读取（三所共用）。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from typing import Any, Callable, Dict, List, Mapping, Optional, Sequence
|
|
|
|
JOURNAL_UPLOAD_TFS: tuple[str, ...] = ("5m", "15m", "1h", "4h")
|
|
JOURNAL_UPLOAD_ALLOWED_EXT = frozenset({".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"})
|
|
|
|
|
|
def journal_upload_field_name(tf: str) -> str:
|
|
return f"screenshot_{tf}"
|
|
|
|
|
|
def _safe_ext(filename: str) -> str:
|
|
ext = os.path.splitext(str(filename or ""))[1].lower()
|
|
return ext if ext in JOURNAL_UPLOAD_ALLOWED_EXT else ".png"
|
|
|
|
|
|
def save_journal_slot_uploads(
|
|
files,
|
|
entry_id: str,
|
|
upload_folder: str,
|
|
*,
|
|
secure_filename_fn: Callable[[str], str],
|
|
) -> List[Dict[str, str]]:
|
|
"""保存四槽位手动截图，返回 [{"tf":"5m","file":"journal_xxx_5m.png"}, ...]。"""
|
|
saved: List[Dict[str, str]] = []
|
|
if not entry_id or not upload_folder:
|
|
return saved
|
|
os.makedirs(upload_folder, exist_ok=True)
|
|
for tf in JOURNAL_UPLOAD_TFS:
|
|
f = files.get(journal_upload_field_name(tf)) if files else None
|
|
if not f or not getattr(f, "filename", None):
|
|
continue
|
|
ext = _safe_ext(f.filename)
|
|
fname = secure_filename_fn(f"journal_{entry_id}_{tf}{ext}")
|
|
if not fname:
|
|
continue
|
|
path = os.path.join(upload_folder, fname)
|
|
f.save(path)
|
|
saved.append({"tf": tf, "file": fname})
|
|
return saved
|
|
|
|
|
|
def images_json_dumps(items: Sequence[Mapping[str, str]]) -> Optional[str]:
|
|
if not items:
|
|
return None
|
|
return json.dumps(list(items), ensure_ascii=False, separators=(",", ":"))
|
|
|
|
|
|
def parse_images_json(raw: Any) -> List[Dict[str, str]]:
|
|
if not raw:
|
|
return []
|
|
if isinstance(raw, list):
|
|
data = raw
|
|
else:
|
|
try:
|
|
data = json.loads(str(raw))
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
return []
|
|
if not isinstance(data, list):
|
|
return []
|
|
out: List[Dict[str, str]] = []
|
|
for item in data:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
tf = str(item.get("tf") or "").strip()
|
|
file = str(item.get("file") or "").strip()
|
|
if file:
|
|
out.append({"tf": tf, "file": file})
|
|
return out
|
|
|
|
|
|
def primary_journal_image(
|
|
manual_images: Sequence[Mapping[str, str]],
|
|
*,
|
|
fallback: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
if manual_images:
|
|
return str(manual_images[0].get("file") or "").strip() or None
|
|
return fallback
|
|
|
|
|
|
def enrich_journal_api_item(item: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""API 输出：解析 images_json，兼容旧单图 image 字段。"""
|
|
images = parse_images_json(item.get("images_json"))
|
|
if not images and item.get("image"):
|
|
images = [{"tf": "", "file": str(item["image"]).strip()}]
|
|
item["images"] = images
|
|
return item
|
|
|
|
|
|
def journal_image_paths(row: Any, upload_folder: str) -> List[str]:
|
|
"""删除 / AI 附图：收集本条复盘所有本地图片路径（去重）。"""
|
|
upload_folder = os.path.abspath(upload_folder or "")
|
|
paths: List[str] = []
|
|
seen = set()
|
|
|
|
def _add(name: Optional[str]) -> None:
|
|
if not name:
|
|
return
|
|
p = os.path.abspath(os.path.join(upload_folder, str(name).strip()))
|
|
if os.path.isfile(p) and p not in seen:
|
|
seen.add(p)
|
|
paths.append(p)
|
|
|
|
try:
|
|
keys = row.keys() if hasattr(row, "keys") else ()
|
|
except Exception:
|
|
keys = ()
|
|
|
|
if "images_json" in keys and row["images_json"]:
|
|
for img in parse_images_json(row["images_json"]):
|
|
_add(img.get("file"))
|
|
if "image" in keys:
|
|
_add(row["image"])
|
|
return paths
|