# -*- coding: utf-8 -*- # your_app/management/commands/import_images_map.py # # Użycie: # python manage.py import_images_map \ # --model houslyspace.AdsNetworkMonitoring \ # --infile /tmp/images_map.csv # # Opcje: # --force -> nadpisuje istniejące images # --dry-run -> pokaż liczby, nie zapisuj # --batch-size N # --images-field images (opcjonalnie) # --url-field url (opcjonalnie) import csv import json from typing import Dict, Tuple, Iterable from django.apps import apps from django.core.management.base import BaseCommand, CommandError from django.db import transaction class Command(BaseCommand): help = ( "Importuje mapę url->images z CSV/JSONL i uzupełnia pole images po dopasowaniu po 'url'.\n" "ID/PK z pliku jest ignorowane." ) def add_arguments(self, parser): parser.add_argument("--model", default="extractly.AdsManual", help="app_label.ModelName (domyślnie: extractly.AdsManual)") parser.add_argument("--images-field", default="images", help="Nazwa pola docelowego (domyślnie: images)") parser.add_argument("--url-field", default="url", help="Nazwa pola klucza (domyślnie: url)") parser.add_argument("--infile", required=True, help="Ścieżka do pliku wejściowego: *.csv lub *.jsonl") parser.add_argument("--batch-size", type=int, default=5000, help="Rozmiar batcha aktualizacji (domyślnie 5000)") parser.add_argument("--force", action="store_true", help="Nadpisuj istniejące images, jeśli różne.") parser.add_argument("--dry-run", action="store_true", help="Pokaż plan aktualizacji, bez zapisu.") parser.add_argument("--strip-query", action="store_true", help="Przy dopasowaniu zignoruj część zapytania (?a=b).") parser.add_argument("--normalize-scheme", choices=["keep", "https", "http"], default="keep", help="Normalizacja schematu URL przy dopasowaniu (domyślnie: keep).") # --- utils --------------------------------------------------------------- @staticmethod def _normalize_url(u: str, strip_query: bool, normalize_scheme: str) -> str: """Prosta normalizacja: trim, opcjonalne usunięcie query, opcjonalna zmiana schematu.""" from urllib.parse import urlparse, urlunparse if not isinstance(u, str): return "" u = u.strip() if not u: return "" p = urlparse(u) scheme = p.scheme if normalize_scheme in ("https", "http"): scheme = normalize_scheme # opcjonalnie usuń query i fragment if strip_query: p = p._replace(query="", fragment="") # zachowaj oryginalną ścieżkę i netloc p = p._replace(scheme=scheme) return urlunparse(p) @staticmethod def _read_csv(path: str) -> Iterable[Tuple[str, str]]: """Zwraca (url, images) z CSV. Kolumny: url, images (pk ignorowane).""" with open(path, "r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) if "url" not in reader.fieldnames or "images" not in reader.fieldnames: raise CommandError("CSV musi mieć kolumny: url, images (pk opcjonalne).") for row in reader: url = (row.get("url") or "").strip() images = (row.get("images") or "").strip() if url and images: yield url, images @staticmethod def _read_jsonl(path: str) -> Iterable[Tuple[str, str]]: """Zwraca (url, images) z JSONL. Oczekuje obiektów z kluczami 'url' i 'images'.""" with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue obj = json.loads(line) url = (obj.get("url") or "").strip() images = (obj.get("images") or "").strip() if url and images: yield url, images # --- main ---------------------------------------------------------------- def handle(self, *args, **opts): model_label = opts["model"] images_field = opts["images_field"] url_field = opts["url_field"] infile = opts["infile"] batch_size = opts["batch_size"] force = bool(opts["force"]) dry = bool(opts["dry_run"]) strip_query = bool(opts["strip_query"]) normalize_scheme = opts["normalize_scheme"] try: Model = apps.get_model(model_label) except LookupError: raise CommandError(f"Model '{model_label}' nie istnieje.") model_fields = {f.name for f in Model._meta.get_fields()} if images_field not in model_fields or url_field not in model_fields: raise CommandError( f"Model {model_label} nie ma pola '{images_field}' lub '{url_field}'." ) # 1) Wczytaj mapę url->images (ostatnie wystąpienie wygrywa) if infile.lower().endswith(".jsonl"): pairs = self._read_jsonl(infile) else: pairs = self._read_csv(infile) mapping: Dict[str, str] = {} total_in = 0 for raw_url, img in pairs: total_in += 1 k = self._normalize_url(raw_url, strip_query=strip_query, normalize_scheme=normalize_scheme) if not k or not img: continue mapping[k] = img self.stdout.write(f"Wczytano z pliku wierszy: {total_in}") self.stdout.write(f"Unikalnych URL po normalizacji: {len(mapping)}") if not mapping: self.stdout.write("Brak danych do importu — kończę.") return # 2) Pobierz obiekty z bazy po url__in (z tą samą normalizacją) # Tu zakładamy, że w DB url jest w tej samej postaci, co k (po normalizacji). urls = list(mapping.keys()) total_updated = 0 total_skipped = 0 total_missing = 0 for i in range(0, len(urls), batch_size): chunk = urls[i : i + batch_size] # DB lookup objs = list( Model.objects.filter(**{f"{url_field}__in": chunk}) .only("pk", url_field, images_field) ) # index po URL z bazy (bez dodatkowej normalizacji — zakładamy spójność) db_by_url = {getattr(o, url_field): o for o in objs} # przygotuj aktualizacje updates = [] for k in chunk: obj = db_by_url.get(k) if not obj: total_missing += 1 continue current = getattr(obj, images_field, None) new_val = mapping[k] # puste/zerowe wartości traktuj jak brak if current in (None, "", {}, []): updates.append((obj.pk, new_val)) elif force and current != new_val: updates.append((obj.pk, new_val)) else: total_skipped += 1 if not updates: continue if dry: total_updated += len(updates) self.stdout.write(f"[DRY] batch {i//batch_size+1}: zaktualizowałbym {len(updates)} rekordów.") continue with transaction.atomic(): for pk, val in updates: Model.objects.filter(pk=pk).update(**{images_field: val}) total_updated += len(updates) self.stdout.write(f"batch {i//batch_size+1}: zaktualizowano {len(updates)}") self.stdout.write(self.style.SUCCESS( f"IMPORT DONE | updated={total_updated} | skipped={total_skipped} | missing_by_url={total_missing}" )) # -*- coding: utf-8 -*- # your_app/management/commands/import_images_map.py # # Użycie: # python manage.py import_images_map \ # --model houslyspace.AdsNetworkMonitoring \ # --infile /tmp/images_map.csv # # Opcje: # --force -> nadpisuje istniejące images # --dry-run -> pokaż liczby, nie zapisuj # --batch-size N # --images-field images (opcjonalnie) # --url-field url (opcjonalnie) import csv import json from typing import Dict, Tuple, Iterable from django.apps import apps from django.core.management.base import BaseCommand, CommandError from django.db import transaction class Command(BaseCommand): help = ( "Importuje mapę url->images z CSV/JSONL i uzupełnia pole images po dopasowaniu po 'url'.\n" "ID/PK z pliku jest ignorowane." ) def add_arguments(self, parser): parser.add_argument("--model", default="extractly.AdsManual", help="app_label.ModelName (domyślnie: extractly.AdsManual)") parser.add_argument("--images-field", default="images", help="Nazwa pola docelowego (domyślnie: images)") parser.add_argument("--url-field", default="url", help="Nazwa pola klucza (domyślnie: url)") parser.add_argument("--infile", required=True, help="Ścieżka do pliku wejściowego: *.csv lub *.jsonl") parser.add_argument("--batch-size", type=int, default=5000, help="Rozmiar batcha aktualizacji (domyślnie 5000)") parser.add_argument("--force", action="store_true", help="Nadpisuj istniejące images, jeśli różne.") parser.add_argument("--dry-run", action="store_true", help="Pokaż plan aktualizacji, bez zapisu.") parser.add_argument("--strip-query", action="store_true", help="Przy dopasowaniu zignoruj część zapytania (?a=b).") parser.add_argument("--normalize-scheme", choices=["keep", "https", "http"], default="keep", help="Normalizacja schematu URL przy dopasowaniu (domyślnie: keep).") # --- utils --------------------------------------------------------------- @staticmethod def _normalize_url(u: str, strip_query: bool, normalize_scheme: str) -> str: """Prosta normalizacja: trim, opcjonalne usunięcie query, opcjonalna zmiana schematu.""" from urllib.parse import urlparse, urlunparse if not isinstance(u, str): return "" u = u.strip() if not u: return "" p = urlparse(u) scheme = p.scheme if normalize_scheme in ("https", "http"): scheme = normalize_scheme # opcjonalnie usuń query i fragment if strip_query: p = p._replace(query="", fragment="") # zachowaj oryginalną ścieżkę i netloc p = p._replace(scheme=scheme) return urlunparse(p) @staticmethod def _read_csv(path: str) -> Iterable[Tuple[str, str]]: """Zwraca (url, images) z CSV. Kolumny: url, images (pk ignorowane).""" with open(path, "r", encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f) if "url" not in reader.fieldnames or "images" not in reader.fieldnames: raise CommandError("CSV musi mieć kolumny: url, images (pk opcjonalne).") for row in reader: url = (row.get("url") or "").strip() images = (row.get("images") or "").strip() if url and images: yield url, images @staticmethod def _read_jsonl(path: str) -> Iterable[Tuple[str, str]]: """Zwraca (url, images) z JSONL. Oczekuje obiektów z kluczami 'url' i 'images'.""" with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue obj = json.loads(line) url = (obj.get("url") or "").strip() images = (obj.get("images") or "").strip() if url and images: yield url, images # --- main ---------------------------------------------------------------- def handle(self, *args, **opts): model_label = opts["model"] images_field = opts["images_field"] url_field = opts["url_field"] infile = opts["infile"] batch_size = opts["batch_size"] force = bool(opts["force"]) dry = bool(opts["dry_run"]) strip_query = bool(opts["strip_query"]) normalize_scheme = opts["normalize_scheme"] try: Model = apps.get_model(model_label) except LookupError: raise CommandError(f"Model '{model_label}' nie istnieje.") model_fields = {f.name for f in Model._meta.get_fields()} if images_field not in model_fields or url_field not in model_fields: raise CommandError( f"Model {model_label} nie ma pola '{images_field}' lub '{url_field}'." ) # 1) Wczytaj mapę url->images (ostatnie wystąpienie wygrywa) if infile.lower().endswith(".jsonl"): pairs = self._read_jsonl(infile) else: pairs = self._read_csv(infile) mapping: Dict[str, str] = {} total_in = 0 for raw_url, img in pairs: total_in += 1 k = self._normalize_url(raw_url, strip_query=strip_query, normalize_scheme=normalize_scheme) if not k or not img: continue mapping[k] = img self.stdout.write(f"Wczytano z pliku wierszy: {total_in}") self.stdout.write(f"Unikalnych URL po normalizacji: {len(mapping)}") if not mapping: self.stdout.write("Brak danych do importu — kończę.") return # 2) Pobierz obiekty z bazy po url__in (z tą samą normalizacją) # Tu zakładamy, że w DB url jest w tej samej postaci, co k (po normalizacji). urls = list(mapping.keys()) total_updated = 0 total_skipped = 0 total_missing = 0 for i in range(0, len(urls), batch_size): chunk = urls[i : i + batch_size] # DB lookup objs = list( Model.objects.filter(**{f"{url_field}__in": chunk}) .only("pk", url_field, images_field) ) # index po URL z bazy (bez dodatkowej normalizacji — zakładamy spójność) db_by_url = {getattr(o, url_field): o for o in objs} # przygotuj aktualizacje updates = [] for k in chunk: obj = db_by_url.get(k) if not obj: total_missing += 1 continue current = getattr(obj, images_field, None) new_val = mapping[k] # puste/zerowe wartości traktuj jak brak if current in (None, "", {}, []): updates.append((obj.pk, new_val)) elif force and current != new_val: updates.append((obj.pk, new_val)) else: total_skipped += 1 if not updates: continue if dry: total_updated += len(updates) self.stdout.write(f"[DRY] batch {i//batch_size+1}: zaktualizowałbym {len(updates)} rekordów.") continue with transaction.atomic(): for pk, val in updates: Model.objects.filter(pk=pk).update(**{images_field: val}) total_updated += len(updates) self.stdout.write(f"batch {i//batch_size+1}: zaktualizowano {len(updates)}") self.stdout.write(self.style.SUCCESS( f"IMPORT DONE | updated={total_updated} | skipped={total_skipped} | missing_by_url={total_missing}" ))