87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
import os
|
|
import re
|
|
|
|
from .config import make_slug
|
|
|
|
|
|
def task_image_path(image_root, task, account=None, suffix="old", ext=".jpg"):
|
|
batch_id = _safe_component(_get(task, "batch_id"), "unknown_batch")
|
|
slug = _account_slug(account, task)
|
|
task_id = _safe_component(_get(task, "id"), "task")
|
|
item_id = _safe_component(_get(task, "item_id"), "item")
|
|
suffix = _safe_component(suffix, "image")
|
|
ext = str(ext or ".jpg")
|
|
if not ext.startswith("."):
|
|
ext = "." + ext
|
|
return os.path.abspath(
|
|
os.path.join(
|
|
str(image_root or "images"),
|
|
batch_id,
|
|
slug,
|
|
f"{task_id}_{item_id}_{suffix}{ext}",
|
|
)
|
|
)
|
|
|
|
|
|
def list_task_cover_candidates(image_root, task, account=None):
|
|
"""Return generated cover candidates for one task from the image directory."""
|
|
|
|
canonical_path = task_image_path(image_root, task, account=account, suffix="new", ext=".jpg")
|
|
directory = os.path.dirname(canonical_path)
|
|
prefix = os.path.splitext(os.path.basename(canonical_path))[0]
|
|
if not os.path.isdir(directory):
|
|
return []
|
|
candidates = []
|
|
for filename in os.listdir(directory):
|
|
full_path = os.path.join(directory, filename)
|
|
if not os.path.isfile(full_path):
|
|
continue
|
|
stem, ext = os.path.splitext(filename)
|
|
if ext.lower() != ".jpg":
|
|
continue
|
|
if stem == prefix or stem.startswith(prefix + "_"):
|
|
candidates.append(os.path.abspath(full_path))
|
|
return sorted(candidates, key=lambda path: _cover_candidate_sort_key(path, prefix))
|
|
|
|
|
|
def _account_slug(account, task):
|
|
slug = _get(account, "slug")
|
|
if slug:
|
|
return _safe_component(slug, "unknown_account")
|
|
alias = _get(task, "alias") or _get(account, "alias") or _get(task, "account_name") or _get(account, "account_name")
|
|
if alias:
|
|
return _safe_component(make_slug(alias), "unknown_account")
|
|
return "unknown_account"
|
|
|
|
|
|
def _cover_candidate_sort_key(path, prefix):
|
|
filename = os.path.basename(path)
|
|
stem, _ext = os.path.splitext(filename)
|
|
if stem == prefix:
|
|
return (0, 0, 0, filename)
|
|
marker = prefix + "_"
|
|
if stem.startswith(marker):
|
|
suffix = stem[len(marker):]
|
|
match = re.fullmatch(r"(\d{14})(?:_(\d+))?", suffix)
|
|
if match:
|
|
timestamp = int(match.group(1))
|
|
collision = int(match.group(2) or 1)
|
|
return (1, -timestamp, -collision, filename)
|
|
return (2, 0, 0, filename)
|
|
|
|
|
|
def _get(obj, name, default=None):
|
|
if obj is None:
|
|
return default
|
|
if isinstance(obj, dict):
|
|
return obj.get(name, default)
|
|
return getattr(obj, name, default)
|
|
|
|
|
|
def _safe_component(value, default):
|
|
text = str(value or "").strip()
|
|
if not text:
|
|
text = str(default)
|
|
safe = "".join(ch if ch.isalnum() or ch in "_-" else "_" for ch in text).strip("_")
|
|
return safe or str(default)
|