mirror of
https://github.com/comfyanonymous/ComfyUI.git
synced 2026-03-05 09:17:37 +08:00
32a6fcf7a8
Some checks are pending
Python Linting / Run Ruff (push) Waiting to run
Python Linting / Run Pylint (push) Waiting to run
Build package / Build Test (3.10) (push) Waiting to run
Build package / Build Test (3.11) (push) Waiting to run
Build package / Build Test (3.12) (push) Waiting to run
Build package / Build Test (3.13) (push) Waiting to run
Build package / Build Test (3.14) (push) Waiting to run
- Reject path separators (/, \, os.sep) in tag components for defense-in-depth - Add comment explaining double-relpath normalization trick - Add _require_assets_feature_enabled decorator returning 503 when disabled - Call asset_seeder.disable() when --enable-assets is not passed - Add iter_chunks to bulk_update_needs_verify, bulk_update_is_missing, and delete_references_by_ids to respect SQLite bind param limits - Fix CacheStateRow.size_bytes NULL coercion (0 -> None) to avoid false needs_verify flags on assets with unknown size - Add PermissionError catch in delete_asset_tags route (403 vs 500) - Add hash-is-None guard in delete_orphaned_seed_asset - Validate from_asset_id in reassign_asset_references - Initialize _prune_first in __init__, remove getattr workaround - Cap error accumulation in _add_error to 200 - Remove confirmed dead code: seed_assets, compute_filename_for_asset, ALLOWED_ROOTS, AssetNotFoundError, SetTagsResult, update_enrichment_level, Asset.to_dict, AssetReference.to_dict, _AssetSeeder.enable Amp-Thread-ID: https://ampcode.com/threads/T-019cb610-1b55-74b6-8dbb-381d73c387c0 Co-authored-by: Amp <amp@ampcode.com>
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
import os
|
|
from datetime import datetime, timezone
|
|
from typing import Sequence
|
|
|
|
|
|
def select_best_live_path(states: Sequence) -> str:
|
|
"""
|
|
Return the best on-disk path among cache states:
|
|
1) Prefer a path that exists with needs_verify == False (already verified).
|
|
2) Otherwise, pick the first path that exists.
|
|
3) Otherwise return empty string.
|
|
"""
|
|
alive = [
|
|
s
|
|
for s in states
|
|
if getattr(s, "file_path", None) and os.path.isfile(s.file_path)
|
|
]
|
|
if not alive:
|
|
return ""
|
|
for s in alive:
|
|
if not getattr(s, "needs_verify", False):
|
|
return s.file_path
|
|
return alive[0].file_path
|
|
|
|
|
|
def escape_sql_like_string(s: str, escape: str = "!") -> tuple[str, str]:
|
|
"""Escapes %, _ and the escape char in a LIKE prefix.
|
|
|
|
Returns (escaped_prefix, escape_char).
|
|
"""
|
|
s = s.replace(escape, escape + escape) # escape the escape char first
|
|
s = s.replace("%", escape + "%").replace("_", escape + "_") # escape LIKE wildcards
|
|
return s, escape
|
|
|
|
|
|
def get_utc_now() -> datetime:
|
|
"""Naive UTC timestamp (no tzinfo). We always treat DB datetimes as UTC."""
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
def normalize_tags(tags: list[str] | None) -> list[str]:
|
|
"""
|
|
Normalize a list of tags by:
|
|
- Stripping whitespace and converting to lowercase.
|
|
- Removing duplicates.
|
|
"""
|
|
return list(dict.fromkeys(t.strip().lower() for t in (tags or []) if (t or "").strip()))
|