Spaces:
Sleeping
Sleeping
| import csv | |
| import re | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Any, Dict, List | |
| EXPORT_DIR = Path("exports") | |
| LATEST_CSV = EXPORT_DIR / "latest.csv" | |
| ARCHIVE_PREFIX = "adm_export_" | |
| CSV_FIELDS = ["Title", "UPC"] | |
| def _clean_title(raw_title: str, studio: str = "") -> str: | |
| if not raw_title: | |
| return "" | |
| t = str(raw_title).strip() | |
| # Remove explicit URLs and common URL-like tokens | |
| t = re.sub(r"https?://\S+", "", t, flags=re.IGNORECASE) | |
| t = re.sub(r"www\.[^\s,]+", "", t, flags=re.IGNORECASE) | |
| # Remove stray backslashes left from escaped quotes in scraped text | |
| t = t.replace("\\", "") | |
| # Remove domain-like tokens (example.com, example.co.uk, cdn.example.com/...) | |
| t = re.sub(r"\b[\w-]+\.(?:com|net|org|io|co|uk|ca|de|jp|au|us|info|biz|online)(?:/[^\s]*)?\b", "", t, flags=re.IGNORECASE) | |
| # Remove "- DVD -" and variants (DVDs, DVD's, DVDS) between title and studio | |
| t = re.sub(r"\s*-\s*DVD(?:s|'s)?\s*-\s*", " - ", t, flags=re.IGNORECASE) | |
| # Remove parenthetical content that contains dvd or urls | |
| t = re.sub(r"\([^)]*(?:dvd|https?://|www\.|/)[^)]*\)", "", t, flags=re.IGNORECASE) | |
| # Remove standalone 'DVD' tokens and common variants like DVD's or DVDs | |
| t = re.sub(r"\bDVD(?:['’]s|s)?\b", "", t, flags=re.IGNORECASE) | |
| # Remove leftover multiple separators and repeated dashes | |
| t = re.sub(r"[-]{2,}", "-", t) | |
| t = re.sub(r"\s*[-–—]\s*", " - ", t) | |
| # Collapse multiple spaces | |
| t = re.sub(r"\s{2,}", " ", t).strip() | |
| # If the studio name is embedded in a longer marketing title, keep only | |
| # the title through the studio token and discard the trailing promo copy. | |
| s = (studio or "").strip() | |
| if s: | |
| studio_match = re.search(re.escape(s), t, flags=re.IGNORECASE) | |
| if studio_match: | |
| t = t[:studio_match.end()].strip() | |
| else: | |
| # Remove occurrences of studio name elsewhere to avoid duplicates, | |
| # then append the canonical studio suffix. | |
| try: | |
| t = re.sub(re.escape(s), "", t, flags=re.IGNORECASE) | |
| except re.error: | |
| pass | |
| t = t.strip(" -\t\n\r") | |
| if t: | |
| t = f"{t} - {s}" | |
| else: | |
| t = s | |
| # Final cleanup: trim and remove stray punctuation at ends | |
| t = t.strip() | |
| t = re.sub(r"^[\-\s,:;]+|[\-\s,:;]+$", "", t) | |
| return t | |
| def _normalize_record(record: Dict[str, Any], fallback_studio: str = "") -> Dict[str, str]: | |
| raw_title = record.get("title") or record.get("Title") or "" | |
| studio = (record.get("studio_name") or record.get("studio") or fallback_studio or "").strip() | |
| title = _clean_title(raw_title, studio) | |
| # Final guarantee: keep only "Movie Title - Studio", drop everything after | |
| parts = title.split(" - ") | |
| if len(parts) > 2: | |
| title = " - ".join(parts[:2]).strip(" -,") | |
| return { | |
| "Title": title, | |
| "UPC": str(record.get("upc") or record.get("UPC") or "").strip(), | |
| } | |
| def save_csv(records: List[Dict[str, Any]], fallback_studio: str = "") -> str: | |
| EXPORT_DIR.mkdir(exist_ok=True) | |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") | |
| archive_path = EXPORT_DIR / f"{ARCHIVE_PREFIX}{timestamp}.csv" | |
| rows = [_normalize_record(record, fallback_studio=fallback_studio) for record in records] | |
| filtered_rows = [] | |
| seen_upc = set() | |
| for row in rows: | |
| upc = row.get("UPC", "").strip() | |
| if not upc: | |
| filtered_rows.append(row) | |
| continue | |
| if upc in seen_upc: | |
| continue | |
| seen_upc.add(upc) | |
| filtered_rows.append(row) | |
| for path in [archive_path, LATEST_CSV]: | |
| with path.open("w", newline="", encoding="utf-8-sig") as handle: | |
| writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS) | |
| writer.writeheader() | |
| writer.writerows(filtered_rows) | |
| return str(LATEST_CSV) | |