73 lines
1.8 KiB
Python
73 lines
1.8 KiB
Python
import re
|
|
from pathlib import Path
|
|
|
|
|
|
TM_ID_PATTERN = re.compile(r"([CS]\d{4})", re.IGNORECASE)
|
|
|
|
|
|
def get_batch_upload_entries(config):
|
|
|
|
bidding_config = config["exchange_bidding"]
|
|
upload_root = Path(bidding_config["upload_input_root"])
|
|
manifest_path = upload_root / "upload_manifest.json"
|
|
|
|
if manifest_path.exists():
|
|
return _load_manifest_entries(manifest_path)
|
|
|
|
return _load_entries_from_filenames(upload_root)
|
|
|
|
|
|
def _load_manifest_entries(manifest_path):
|
|
|
|
import json
|
|
|
|
with manifest_path.open("r", encoding="utf-8") as file:
|
|
payload = json.load(file)
|
|
|
|
entries = []
|
|
for item in payload.get("entries", []):
|
|
entries.append(
|
|
{
|
|
"tm_id": item["tm_id"],
|
|
"tac_id": item["tac_id"],
|
|
"upload_path": str((manifest_path.parent / item["file_name"]).resolve()),
|
|
"file_name": item["file_name"],
|
|
}
|
|
)
|
|
return entries
|
|
|
|
|
|
def _load_entries_from_filenames(upload_root):
|
|
|
|
entries = []
|
|
for path in sorted(upload_root.iterdir()):
|
|
if not path.is_file() or path.name.startswith("."):
|
|
continue
|
|
|
|
tac_id = path.stem.upper()
|
|
tm_id = _derive_tm_id_from_tac(tac_id)
|
|
entries.append(
|
|
{
|
|
"tm_id": tm_id,
|
|
"tac_id": tac_id,
|
|
"upload_path": str(path.resolve()),
|
|
"file_name": path.name,
|
|
}
|
|
)
|
|
|
|
if not entries:
|
|
raise FileNotFoundError(f"No upload files were found in {upload_root}")
|
|
|
|
return entries
|
|
|
|
|
|
def _derive_tm_id_from_tac(tac_id):
|
|
|
|
match = TM_ID_PATTERN.search(tac_id)
|
|
if not match:
|
|
raise ValueError(
|
|
f"Could not derive TM ID from TAC/file name '{tac_id}'. "
|
|
"Use a manifest file for custom mapping."
|
|
)
|
|
return match.group(1).upper()
|