crypto_monitor/lib/instance/journal_images_lib.py

"""复盘记录:多周期截图上传,存储与读取(三所共用)."""
from __future__ import annotations

import json
import os
import re
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"})
_JOURNAL_DRAFT_ID_RE = re.compile(r"^[a-f0-9]{32}$")
_JOURNAL_SLOT_FILE_RE = re.compile(
    r"^journal_([a-f0-9]{32})_(5m|15m|1h|4h)\.(png|jpg|jpeg|webp|gif|bmp)$",
    re.I,
)


def journal_upload_field_name(tf: str) -> str:
    return f"screenshot_{tf}"


def uploaded_screenshot_field_name(tf: str) -> str:
    return f"uploaded_screenshot_{tf}"


def normalize_journal_draft_id(raw: Any) -> Optional[str]:
    s = str(raw or "").strip().lower()
    if _JOURNAL_DRAFT_ID_RE.match(s):
        return s
    return None


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 build_journal_slot_filename(
    entry_id: str,
    tf: str,
    ext: str,
    *,
    secure_filename_fn: Callable[[str], str],
) -> str:
    ext = ext if ext.startswith(".") else f".{ext}"
    ext = _safe_ext(f"x{ext}")
    fname = secure_filename_fn(f"journal_{entry_id}_{tf}{ext}")
    return fname or ""


def is_valid_preuploaded_journal_file(filename: str, entry_id: str, tf: str) -> bool:
    fn = os.path.basename(str(filename or "").strip())
    if not fn or fn != str(filename or "").strip():
        return False
    m = _JOURNAL_SLOT_FILE_RE.match(fn)
    if not m:
        return False
    return m.group(1) == entry_id.lower() and m.group(2) == tf


def save_journal_slot_file(
    file,
    entry_id: str,
    tf: str,
    upload_folder: str,
    *,
    secure_filename_fn: Callable[[str], str],
) -> Optional[Dict[str, str]]:
    if tf not in JOURNAL_UPLOAD_TFS or not entry_id or not upload_folder:
        return None
    if not file or not getattr(file, "filename", None):
        return None
    ext = _safe_ext(file.filename)
    fname = build_journal_slot_filename(
        entry_id, tf, ext, secure_filename_fn=secure_filename_fn
    )
    if not fname:
        return None
    os.makedirs(upload_folder, exist_ok=True)
    path = os.path.join(upload_folder, fname)
    file.save(path)
    return {"tf": tf, "file": fname}


def collect_journal_slot_images(
    form,
    files,
    entry_id: str,
    upload_folder: str,
    *,
    secure_filename_fn: Callable[[str], str],
) -> List[Dict[str, str]]:
    """优先使用即时上传 hidden 字段;否则回退到表单 multipart."""
    saved: List[Dict[str, str]] = []
    if not entry_id or not upload_folder:
        return saved
    for tf in JOURNAL_UPLOAD_TFS:
        pre = ""
        if form is not None:
            pre = str(form.get(uploaded_screenshot_field_name(tf)) or "").strip()
        if pre and is_valid_preuploaded_journal_file(pre, entry_id, tf):
            path = os.path.join(upload_folder, os.path.basename(pre))
            if os.path.isfile(path):
                saved.append({"tf": tf, "file": os.path.basename(pre)})
                continue
        f = files.get(journal_upload_field_name(tf)) if files else None
        item = save_journal_slot_file(
            f,
            entry_id,
            tf,
            upload_folder,
            secure_filename_fn=secure_filename_fn,
        )
        if item:
            saved.append(item)
    return saved


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"}, ...]."""
    return collect_journal_slot_images(
        None,
        files,
        entry_id,
        upload_folder,
        secure_filename_fn=secure_filename_fn,
    )


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