Reformat code
This commit is contained in:
@@ -8,51 +8,47 @@ import toml
|
||||
# Default Schema
|
||||
# This defines the "First Class" feel of the app, but is fully overridable.
|
||||
DEFAULT_CONFIG: dict[str, Any] = {
|
||||
"sidecar": {
|
||||
"extension": ".xmp"
|
||||
},
|
||||
"defaults": {
|
||||
"time_increment": 60
|
||||
},
|
||||
"mappings": {
|
||||
"author": {
|
||||
"flags": [
|
||||
"-Artist={value}",
|
||||
"-Creator={value}",
|
||||
"-By-line={value}",
|
||||
"-Credit={value}",
|
||||
"-CopyrightNotice=© {year} {value}",
|
||||
"-Copyright=© {year} {value}"
|
||||
'sidecar': {'extension': '.xmp'},
|
||||
'defaults': {'time_increment': 60},
|
||||
'mappings': {
|
||||
'author': {
|
||||
'flags': [
|
||||
'-Artist={value}',
|
||||
'-Creator={value}',
|
||||
'-By-line={value}',
|
||||
'-Credit={value}',
|
||||
'-CopyrightNotice=© {year} {value}',
|
||||
'-Copyright=© {year} {value}',
|
||||
],
|
||||
"prompt": True,
|
||||
"help": "Name of the photographer"
|
||||
'prompt': True,
|
||||
'help': 'Name of the photographer',
|
||||
},
|
||||
"lab": {
|
||||
"flags": ["-XMP:DevelopedBy={value}"],
|
||||
"prompt": True,
|
||||
"help": "Lab name"
|
||||
'lab': {
|
||||
'flags': ['-XMP:DevelopedBy={value}'],
|
||||
'prompt': True,
|
||||
'help': 'Lab name',
|
||||
},
|
||||
"make": {
|
||||
"flags": ["-Make={value}"],
|
||||
"prompt": True,
|
||||
"help": "Camera make"
|
||||
'make': {
|
||||
'flags': ['-Make={value}'],
|
||||
'prompt': True,
|
||||
'help': 'Camera make',
|
||||
},
|
||||
"model": {
|
||||
"flags": ["-Model={value}"],
|
||||
"prompt": True,
|
||||
"help": "Camera model"
|
||||
'model': {
|
||||
'flags': ['-Model={value}'],
|
||||
'prompt': True,
|
||||
'help': 'Camera model',
|
||||
},
|
||||
"lens": {
|
||||
"flags": ["-LensModel={value}", "-Lens={value}"],
|
||||
"prompt": True,
|
||||
"help": "Lens model"
|
||||
'lens': {
|
||||
'flags': ['-LensModel={value}', '-Lens={value}'],
|
||||
'prompt': True,
|
||||
'help': 'Lens model',
|
||||
},
|
||||
"film": {
|
||||
"flags": ["-UserComment={value}", "-XMP:Description={value}"],
|
||||
"prompt": False,
|
||||
"help": "Film stock"
|
||||
}
|
||||
}
|
||||
'film': {
|
||||
'flags': ['-UserComment={value}', '-XMP:Description={value}'],
|
||||
'prompt': False,
|
||||
'help': 'Film stock',
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -25,22 +25,21 @@ class Executor:
|
||||
def __init__(self, config: dict[str, Any]) -> None:
|
||||
self.config = config
|
||||
self.mappings: dict[str, Any] = config.get('mappings', {})
|
||||
self.sidecar_ext: str = config.get(
|
||||
'sidecar', {}).get('extension', '.xmp')
|
||||
self.sidecar_ext: str = config.get('sidecar', {}).get(
|
||||
'extension', '.xmp'
|
||||
)
|
||||
if not self.sidecar_ext.startswith('.'):
|
||||
self.sidecar_ext = f".{self.sidecar_ext}"
|
||||
self.sidecar_ext = f'.{self.sidecar_ext}'
|
||||
|
||||
def create_tasks(
|
||||
self, files: list[str],
|
||||
self,
|
||||
files: list[str],
|
||||
resolved_values: dict[str, str],
|
||||
options: Namespace
|
||||
options: Namespace,
|
||||
) -> list[Task]:
|
||||
# Filter supported files
|
||||
extensions = ['.jpg', '.jpeg', '.tif', '.tiff']
|
||||
valid_files = [
|
||||
f for f in files
|
||||
if Path(f).suffix.lower() in extensions
|
||||
]
|
||||
valid_files = [f for f in files if Path(f).suffix.lower() in extensions]
|
||||
|
||||
if not valid_files:
|
||||
return []
|
||||
@@ -54,7 +53,7 @@ class Executor:
|
||||
for i, f in enumerate(valid_files):
|
||||
# Calculate timestamp
|
||||
ts_dt = base_dt + datetime.timedelta(seconds=i * time_increment)
|
||||
timestamp_str = ts_dt.strftime("%Y:%m:%d %H:%M:%S")
|
||||
timestamp_str = ts_dt.strftime('%Y:%m:%d %H:%M:%S')
|
||||
|
||||
# Determine file targets (Sidecar logic)
|
||||
target_path, sidecar_source = self._determine_paths(
|
||||
@@ -63,10 +62,7 @@ class Executor:
|
||||
|
||||
# Build Command
|
||||
cmd = self._build_cmd(
|
||||
target_path,
|
||||
resolved_values,
|
||||
timestamp_str,
|
||||
sidecar_source
|
||||
target_path, resolved_values, timestamp_str, sidecar_source
|
||||
)
|
||||
tasks.append(Task(f, tuple(cmd), timestamp_str))
|
||||
|
||||
@@ -80,7 +76,7 @@ class Executor:
|
||||
return self._run_exiftool(list(task.command), dry_run)
|
||||
|
||||
def _determine_paths(
|
||||
self, original_file: Path, embed: bool
|
||||
self, original_file: Path, embed: bool
|
||||
) -> tuple[Path, Path | None]:
|
||||
"""
|
||||
Returns (target_path, sidecar_source_if_needed)
|
||||
@@ -102,18 +98,18 @@ class Executor:
|
||||
file_path: Path,
|
||||
field_values: dict[str, str],
|
||||
timestamp_str: str,
|
||||
sidecar_source: Path | None = None
|
||||
sidecar_source: Path | None = None,
|
||||
) -> list[str]:
|
||||
current_year = datetime.datetime.now().year
|
||||
|
||||
# Core setup
|
||||
cmd = [
|
||||
"exiftool",
|
||||
"-overwrite_original",
|
||||
f"-DateTimeOriginal={timestamp_str}",
|
||||
f"-CreateDate={timestamp_str}",
|
||||
"-WebStatement=",
|
||||
"-CreatorWorkURL="
|
||||
'exiftool',
|
||||
'-overwrite_original',
|
||||
f'-DateTimeOriginal={timestamp_str}',
|
||||
f'-CreateDate={timestamp_str}',
|
||||
'-WebStatement=',
|
||||
'-CreatorWorkURL=',
|
||||
]
|
||||
|
||||
# Add mapped fields
|
||||
@@ -130,16 +126,15 @@ class Executor:
|
||||
# Ensure flags is a list (just in case)
|
||||
if isinstance(flags, list):
|
||||
for flag in flags:
|
||||
safe_flag = flag.replace(
|
||||
'{value}',
|
||||
str(val)).replace('{year}', str(current_year)
|
||||
safe_flag = flag.replace('{value}', str(val)).replace(
|
||||
'{year}', str(current_year)
|
||||
)
|
||||
cmd.append(safe_flag)
|
||||
|
||||
# Sidecar handling
|
||||
if sidecar_source:
|
||||
# -srcfile SOURCE TARGET
|
||||
cmd.append("-srcfile")
|
||||
cmd.append('-srcfile')
|
||||
cmd.append(str(file_path))
|
||||
cmd.append(str(sidecar_source))
|
||||
else:
|
||||
@@ -150,18 +145,18 @@ class Executor:
|
||||
def _run_exiftool(self, cmd: list[str], dry_run: bool) -> tuple[bool, str]:
|
||||
if dry_run:
|
||||
safe_cmd = shlex.join(cmd)
|
||||
return True, f"[DRY RUN] {safe_cmd}"
|
||||
return True, f'[DRY RUN] {safe_cmd}'
|
||||
|
||||
try:
|
||||
subprocess.run( # noqa: S603
|
||||
cmd,
|
||||
check=True,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
return True, "Updated"
|
||||
return True, 'Updated'
|
||||
except subprocess.CalledProcessError as e:
|
||||
return False, f"Error: {e}"
|
||||
return False, f'Error: {e}'
|
||||
except FileNotFoundError:
|
||||
return False, "Error: 'exiftool' not found. Please install it."
|
||||
|
||||
@@ -170,6 +165,6 @@ class Executor:
|
||||
# Should be handled by resolver/validator, but safe fallback
|
||||
return datetime.datetime.now()
|
||||
dt_str = dt_str.strip()
|
||||
if " " in dt_str:
|
||||
return datetime.datetime.strptime(dt_str, "%Y-%m-%d %H:%M:%S")
|
||||
return datetime.datetime.strptime(dt_str, "%Y-%m-%d")
|
||||
if ' ' in dt_str:
|
||||
return datetime.datetime.strptime(dt_str, '%Y-%m-%d %H:%M:%S')
|
||||
return datetime.datetime.strptime(dt_str, '%Y-%m-%d')
|
||||
|
||||
@@ -66,7 +66,7 @@ def parse_args() -> argparse.Namespace:
|
||||
'--time-increment',
|
||||
type=int,
|
||||
default=None,
|
||||
help='Time increment in seconds between images.'
|
||||
help='Time increment in seconds between images.',
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
@@ -80,7 +80,7 @@ def parse_args() -> argparse.Namespace:
|
||||
parser.add_argument(
|
||||
'--dry-run',
|
||||
action='store_true',
|
||||
help='Show what would be changed without modifying files.'
|
||||
help='Show what would be changed without modifying files.',
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
@@ -99,13 +99,13 @@ def parse_args() -> argparse.Namespace:
|
||||
parser.add_argument(
|
||||
'--init-config',
|
||||
action='store_true',
|
||||
help='Create a default config file (if none exists) and exit.'
|
||||
help='Create a default config file (if none exists) and exit.',
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
'--no-interaction',
|
||||
action='store_true',
|
||||
help='Do not prompt for missing fields (skip them if missing).'
|
||||
help='Do not prompt for missing fields (skip them if missing).',
|
||||
)
|
||||
|
||||
# --- UI Options ---
|
||||
@@ -114,7 +114,7 @@ def parse_args() -> argparse.Namespace:
|
||||
'--no-tui',
|
||||
dest='simple_ui',
|
||||
action='store_true',
|
||||
help='Force the simple text interface.'
|
||||
help='Force the simple text interface.',
|
||||
)
|
||||
|
||||
return parser.parse_args()
|
||||
@@ -247,11 +247,11 @@ def main() -> None:
|
||||
try:
|
||||
tasks = executor.create_tasks(args.files, resolved_values, args)
|
||||
except ValueError as e:
|
||||
print(f"Error creating tasks: {e}")
|
||||
print(f'Error creating tasks: {e}')
|
||||
sys.exit(1)
|
||||
|
||||
if not tasks:
|
||||
print("No valid image files found to process.")
|
||||
print('No valid image files found to process.')
|
||||
sys.exit(0)
|
||||
|
||||
# Create Runner (Controller)
|
||||
@@ -259,7 +259,7 @@ def main() -> None:
|
||||
executor=executor,
|
||||
tasks=tasks,
|
||||
dry_run=args.dry_run,
|
||||
workers=args.workers or 1
|
||||
workers=args.workers or 1,
|
||||
)
|
||||
|
||||
# Select UI (View)
|
||||
|
||||
@@ -92,11 +92,13 @@ class PhysicsState:
|
||||
eta: float | None = None
|
||||
percent: float = 0.0
|
||||
|
||||
def __iter__(self) -> tuple[
|
||||
float | None,
|
||||
float | None,
|
||||
float,
|
||||
]:
|
||||
def __iter__(
|
||||
self,
|
||||
) -> tuple[
|
||||
float | None,
|
||||
float | None,
|
||||
float,
|
||||
]:
|
||||
return iter(astuple(self))
|
||||
|
||||
|
||||
@@ -225,13 +227,13 @@ class ProgressBar:
|
||||
return self.spinner_chars[spinner_idx]
|
||||
|
||||
def _render_stats(
|
||||
self,
|
||||
current: int,
|
||||
total: int,
|
||||
rate: float,
|
||||
eta: float,
|
||||
percent: float,
|
||||
) -> str:
|
||||
self,
|
||||
current: int,
|
||||
total: int,
|
||||
rate: float,
|
||||
eta: float,
|
||||
percent: float,
|
||||
) -> str:
|
||||
eta_s = format_time(int(eta)) if eta is not None else '???'
|
||||
|
||||
rate_val = format_si(rate or 0, sigfigs=2, width=3).strip()
|
||||
|
||||
@@ -12,8 +12,8 @@ class ValueResolver:
|
||||
self.defaults: dict[str, Any] = config.get('defaults', {})
|
||||
|
||||
def resolve(
|
||||
self, cli_args: dict[str, Any], interactive: bool = True
|
||||
) -> dict[str, str]:
|
||||
self, cli_args: dict[str, Any], interactive: bool = True
|
||||
) -> dict[str, str]:
|
||||
"""
|
||||
Resolves the final values for all fields.
|
||||
|
||||
@@ -46,7 +46,8 @@ class ValueResolver:
|
||||
fields_to_prompt: list[str] = []
|
||||
for field_name, schema in self.mappings.items():
|
||||
# Check if prompt is requested
|
||||
if (isinstance(schema, dict)
|
||||
if (
|
||||
isinstance(schema, dict)
|
||||
and schema.get('prompt', False)
|
||||
# Check if we already have a value
|
||||
and (field_name not in resolved or not resolved[field_name])
|
||||
@@ -61,7 +62,7 @@ class ValueResolver:
|
||||
for field in fields_to_prompt:
|
||||
self._prompt_user(field, resolved)
|
||||
except KeyboardInterrupt:
|
||||
print("\nInterrupted. Exiting.")
|
||||
print('\nInterrupted. Exiting.')
|
||||
sys.exit(1)
|
||||
|
||||
# Remove any fields that are still None (but keep empty strings)
|
||||
@@ -69,22 +70,22 @@ class ValueResolver:
|
||||
return {k: str(v) for k, v in resolved.items() if v is not None}
|
||||
|
||||
def _prompt_user(
|
||||
self, field_name: str, resolved_dict: dict[str, Any]
|
||||
self, field_name: str, resolved_dict: dict[str, Any]
|
||||
) -> None:
|
||||
"""
|
||||
Helper to prompt a single field.
|
||||
"""
|
||||
schema = self.mappings.get(field_name, {})
|
||||
help_text = (
|
||||
schema.get('help', field_name)
|
||||
if isinstance(schema, dict)
|
||||
else field_name
|
||||
schema.get('help', field_name)
|
||||
if isinstance(schema, dict)
|
||||
else field_name
|
||||
)
|
||||
|
||||
# We capitalize the field name for the prompt label if help text matches
|
||||
# name
|
||||
label = help_text
|
||||
|
||||
val = input(f"{label} (Optional): ").strip()
|
||||
val = input(f'{label} (Optional): ').strip()
|
||||
if val:
|
||||
resolved_dict[field_name] = val
|
||||
|
||||
@@ -64,9 +64,7 @@ def _validate_args(
|
||||
)
|
||||
|
||||
|
||||
def _get_prefixes(
|
||||
prefixes: PrefixTable | str, ascii_mode: bool
|
||||
) -> PrefixTable:
|
||||
def _get_prefixes(prefixes: PrefixTable | str, ascii_mode: bool) -> PrefixTable:
|
||||
if isinstance(prefixes, str):
|
||||
if prefixes != 'SI':
|
||||
raise ValueError("prefixes must be a dict or 'SI'")
|
||||
|
||||
@@ -4,14 +4,15 @@
|
||||
|
||||
intervals = (
|
||||
('yr', 31536000), # 60 * 60 * 24 * 365
|
||||
('mo', 2592000), # 60 * 60 * 24 * 30
|
||||
('mo', 2592000), # 60 * 60 * 24 * 30
|
||||
('wks', 604800), # 60 * 60 * 24 * 7
|
||||
('d', 86400), # 60 * 60 * 24
|
||||
('d', 86400), # 60 * 60 * 24
|
||||
('h', 3600), # 60 * 60
|
||||
('m', 60),
|
||||
('s', 1),
|
||||
)
|
||||
|
||||
|
||||
def format_time(seconds: float, granularity: int = 2) -> str:
|
||||
result = []
|
||||
|
||||
@@ -21,5 +22,5 @@ def format_time(seconds: float, granularity: int = 2) -> str:
|
||||
seconds -= value * count
|
||||
if value == 1:
|
||||
name = name.rstrip('s')
|
||||
result.append(f"{value}{name}")
|
||||
result.append(f'{value}{name}')
|
||||
return ' '.join(result[:granularity])
|
||||
|
||||
Reference in New Issue
Block a user