Spaces:
Sleeping
Sleeping
| import os | |
| import re | |
| import io | |
| import time | |
| import math | |
| import hashlib | |
| import zipfile | |
| import tempfile | |
| from pathlib import Path | |
| from datetime import datetime, timezone | |
| from urllib.parse import urlparse, parse_qs, unquote, urljoin | |
| import requests | |
| from requests.adapters import HTTPAdapter | |
| from concurrent.futures import ThreadPoolExecutor | |
| from bs4 import BeautifulSoup | |
| import pandas as pd | |
| import numpy as np | |
| from PIL import Image, ExifTags | |
| import matplotlib | |
| matplotlib.use("Agg") | |
| import matplotlib.pyplot as plt | |
| import gradio as gr | |
| # ============================================================ | |
| # CONFIGURATION | |
| # ============================================================ | |
| APP_TITLE = "Web & Flickr Image Downloader + Metadata" | |
| MAX_IMAGES_HARD_LIMIT = 500 | |
| REQUEST_TIMEOUT = 30 | |
| IMAGE_TIMEOUT = 45 | |
| HEADERS = { | |
| "User-Agent": ( | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " | |
| "AppleWebKit/537.36 (KHTML, like Gecko) " | |
| "Chrome/140.0 Safari/537.36" | |
| ), | |
| "Accept": ( | |
| "text/html,application/xhtml+xml,application/xml;q=0.9," | |
| "image/avif,image/webp,image/apng,*/*;q=0.8" | |
| ), | |
| "Accept-Language": "en-US,en;q=0.9,fr;q=0.8", | |
| } | |
| BASE_DIR = Path(tempfile.gettempdir()) / "gradio_web_image_downloader" | |
| BASE_DIR.mkdir(parents=True, exist_ok=True) | |
| MAX_WORKERS = 8 # parallel requests (raise for speed, lower if Flickr starts refusing) | |
| session = requests.Session() | |
| session.headers.update(HEADERS) | |
| _adapter = HTTPAdapter(pool_connections=MAX_WORKERS * 2, pool_maxsize=MAX_WORKERS * 2) | |
| session.mount("https://", _adapter) | |
| session.mount("http://", _adapter) | |
| EXIF_TAGS = dict(ExifTags.TAGS) | |
| GPS_TAGS = dict(ExifTags.GPSTAGS) | |
| FLICKR_PHOTO_RE = re.compile(r"flickr\.com/photos/([^/?#]+)/(?:with/)?(\d{6,})") | |
| # ============================================================ | |
| # GENERIC HELPERS | |
| # ============================================================ | |
| def clean_url(url): | |
| url = str(url or "").strip() | |
| if url and not url.startswith(("http://", "https://")): | |
| url = "https://" + url | |
| return url | |
| def valid_http_url(url): | |
| try: | |
| p = urlparse(url) | |
| return p.scheme in ("http", "https") and bool(p.netloc) | |
| except Exception: | |
| return False | |
| def is_flickr_url(url): | |
| try: | |
| host = urlparse(url).netloc.lower() | |
| return host == "flickr.com" or host.endswith(".flickr.com") | |
| except Exception: | |
| return False | |
| def extract_flickr_photo_id(url): | |
| """Return (username, photo_id) for /photos/<user>/<id> or /photos/<user>/with/<id>.""" | |
| m = FLICKR_PHOTO_RE.search(url) | |
| return (m.group(1), m.group(2)) if m else (None, None) | |
| def safe_filename(value, default="image"): | |
| value = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", str(value or default)).strip(" .") | |
| return (value or default)[:180] | |
| def make_hash(value): | |
| return hashlib.sha1(value.encode("utf-8", errors="ignore")).hexdigest()[:12] | |
| CONTENT_TYPE_EXT = { | |
| "image/jpeg": ".jpg", "image/jpg": ".jpg", "image/png": ".png", | |
| "image/webp": ".webp", "image/gif": ".gif", "image/bmp": ".bmp", | |
| "image/tiff": ".tif", "image/avif": ".avif", | |
| } | |
| FORMAT_EXT = { | |
| "jpeg": ".jpg", "jpg": ".jpg", "png": ".png", "webp": ".webp", | |
| "gif": ".gif", "bmp": ".bmp", "tiff": ".tif", "avif": ".avif", | |
| } | |
| def make_item(**kw): | |
| """Uniform candidate record used by every source.""" | |
| item = { | |
| "image_url": "", "page_url": "", "title": "", "owner": "", | |
| "photo_id": "", "width": 0, "height": 0, "source": "", | |
| "license": "", "date_taken": "", "date_uploaded": "", | |
| "views": "", "tags": "", "description": "", "alt": "", | |
| } | |
| item.update(kw) | |
| return item | |
| # ============================================================ | |
| # DATE HELPERS | |
| # ============================================================ | |
| def unix_to_str(value): | |
| """Unix seconds (int or str) -> 'YYYY-MM-DD HH:MM:SS' (UTC).""" | |
| try: | |
| return datetime.fromtimestamp(int(value), tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S") | |
| except Exception: | |
| return "" | |
| def parse_date(value): | |
| if value is None: | |
| return pd.NaT | |
| if isinstance(value, pd.Timestamp): | |
| return value | |
| if isinstance(value, str) and re.fullmatch(r"\d{9,13}", value.strip()): | |
| value = int(value.strip()) # Flickr returns Unix timestamps as strings | |
| if isinstance(value, (int, float)): | |
| if value > 100000000000: | |
| return pd.to_datetime(value, unit="ms", errors="coerce") | |
| if value > 1000000000: | |
| return pd.to_datetime(value, unit="s", errors="coerce") | |
| value = str(value).strip() | |
| if not value: | |
| return pd.NaT | |
| if re.match(r"^\d{4}:\d{2}:\d{2}", value): # EXIF format | |
| dt = pd.to_datetime(value, format="%Y:%m:%d %H:%M:%S", errors="coerce") | |
| if not pd.isna(dt): | |
| return dt | |
| dt = pd.to_datetime(value, errors="coerce") | |
| if not pd.isna(dt) and getattr(dt, "tzinfo", None) is not None: | |
| dt = dt.tz_convert("UTC").tz_localize(None) | |
| return dt | |
| def iso_or_empty(value): | |
| dt = parse_date(value) | |
| return "" if pd.isna(dt) else dt.strftime("%Y-%m-%d %H:%M:%S") | |
| # ============================================================ | |
| # EXIF | |
| # ============================================================ | |
| def _num(x): | |
| """Convert PIL rational / tuple to float.""" | |
| if isinstance(x, tuple) and len(x) == 2: | |
| return float(x[0]) / float(x[1]) if x[1] else 0.0 | |
| return float(x) | |
| def gps_to_decimal(value, ref=None): | |
| try: | |
| if not value or len(value) != 3: | |
| return None | |
| d, m, s = (_num(v) for v in value) | |
| dec = d + m / 60 + s / 3600 | |
| return -dec if ref in ("S", "W") else dec | |
| except Exception: | |
| return None | |
| def extract_exif_metadata(image): | |
| md = {k: "" for k in [ | |
| "exif_make", "exif_model", "exif_software", "exif_datetime", | |
| "exif_datetime_original", "exif_datetime_digitized", "exif_orientation", | |
| "exif_lens_model", "exif_focal_length", "exif_iso", "exif_exposure_time", | |
| "exif_f_number", "exif_gps_latitude", "exif_gps_longitude", "exif_gps_altitude", | |
| ]} | |
| try: | |
| exif = image.getexif() | |
| if not exif: | |
| return md | |
| # Base IFD + Exif sub-IFD (where DateTimeOriginal, ISO, etc. actually live) | |
| decoded = {EXIF_TAGS.get(k, str(k)): v for k, v in exif.items()} | |
| try: | |
| for k, v in exif.get_ifd(0x8769).items(): | |
| decoded[EXIF_TAGS.get(k, str(k))] = v | |
| except Exception: | |
| pass | |
| g = lambda k: str(decoded.get(k, "")) | |
| md.update({ | |
| "exif_make": g("Make"), "exif_model": g("Model"), | |
| "exif_software": g("Software"), "exif_datetime": g("DateTime"), | |
| "exif_datetime_original": g("DateTimeOriginal"), | |
| "exif_datetime_digitized": g("DateTimeDigitized"), | |
| "exif_orientation": g("Orientation"), "exif_lens_model": g("LensModel"), | |
| "exif_iso": str(decoded.get("ISOSpeedRatings", decoded.get("PhotographicSensitivity", ""))), | |
| "exif_exposure_time": g("ExposureTime"), "exif_f_number": g("FNumber"), | |
| }) | |
| if decoded.get("FocalLength") is not None: | |
| try: | |
| md["exif_focal_length"] = round(_num(decoded["FocalLength"]), 2) | |
| except Exception: | |
| md["exif_focal_length"] = g("FocalLength") | |
| try: | |
| gps = {GPS_TAGS.get(k, str(k)): v for k, v in exif.get_ifd(0x8825).items()} | |
| except Exception: | |
| gps = {} | |
| lat = gps_to_decimal(gps.get("GPSLatitude"), gps.get("GPSLatitudeRef")) | |
| lon = gps_to_decimal(gps.get("GPSLongitude"), gps.get("GPSLongitudeRef")) | |
| if lat is not None: | |
| md["exif_gps_latitude"] = round(lat, 7) | |
| if lon is not None: | |
| md["exif_gps_longitude"] = round(lon, 7) | |
| if gps.get("GPSAltitude") is not None: | |
| try: | |
| md["exif_gps_altitude"] = round(_num(gps["GPSAltitude"]), 2) | |
| except Exception: | |
| md["exif_gps_altitude"] = str(gps["GPSAltitude"]) | |
| except Exception: | |
| pass | |
| return md | |
| # ============================================================ | |
| # FLICKR | |
| # ============================================================ | |
| def flickr_photo_via_html(username, photo_id, page_url): | |
| """Read the publication date and the largest image from the public photo page.""" | |
| r = session.get(page_url, timeout=REQUEST_TIMEOUT) | |
| r.raise_for_status() | |
| html = r.text.replace("\\/", "/") | |
| soup = BeautifulSoup(html, "html.parser") | |
| date_uploaded = "" | |
| for pattern, is_unix in [ | |
| (r'"datePosted"\s*:\s*"?(\d{9,10})"?', True), | |
| (r'"date_?[Uu]ploaded"\s*:\s*"?(\d{9,10})"?', True), | |
| (r'"datePublished"\s*:\s*"([^"]+)"', False), | |
| ]: | |
| m = re.search(pattern, html) | |
| if m: | |
| date_uploaded = unix_to_str(m.group(1)) if is_unix else iso_or_empty(m.group(1)) | |
| if date_uploaded: | |
| break | |
| m = re.search(r'"dateTaken"\s*:\s*"([^"]+)"', html) | |
| date_taken = m.group(1) if m else "" | |
| rank = ["o", "k", "h", "b", "c", "z"] | |
| found = {} | |
| for m in re.finditer( | |
| r'((?:https?:)?//live\.staticflickr\.com/\d+/' + re.escape(photo_id) | |
| + r'_[0-9a-f]+(?:_([a-z0-9]))?\.(?:jpg|png|gif))', html): | |
| found[m.group(2) or "m"] = m.group(1) | |
| image_url = next((found[k] for k in rank if k in found), None) | |
| if not image_url and found: | |
| image_url = list(found.values())[-1] | |
| if not image_url: | |
| og = soup.find("meta", attrs={"property": "og:image"}) | |
| image_url = og.get("content", "") if og else "" | |
| if image_url.startswith("//"): | |
| image_url = "https:" + image_url | |
| if not image_url: | |
| raise RuntimeError("Could not find the image on this Flickr page.") | |
| og_title = soup.find("meta", attrs={"property": "og:title"}) | |
| return make_item( | |
| image_url=image_url, page_url=page_url, | |
| title=og_title.get("content", "") if og_title else "", | |
| owner=username, photo_id=photo_id, source="Flickr HTML (photo page)", | |
| date_taken=date_taken, date_uploaded=date_uploaded, | |
| ) | |
| def flickr_photo_candidate(url): | |
| username, photo_id = extract_flickr_photo_id(url) | |
| if not photo_id: | |
| return None | |
| page_url = f"https://www.flickr.com/photos/{username}/{photo_id}/" | |
| return flickr_photo_via_html(username, photo_id, page_url) | |
| STREAM_STATS = {} | |
| def flickr_stream_via_html(username, max_images, first, max_pages=100): | |
| """Read the photostream pages (newest first), then each photo page. | |
| Collects a few spare IDs so failed photo pages can be replaced.""" | |
| stats = STREAM_STATS | |
| stats.clear() | |
| stats.update(pages=0, ids_per_page=[], failed=0, errors=[], stop="") | |
| wanted = max_images * 2 # spare IDs replace photo pages that fail to load | |
| def parse_ids(html): | |
| html = html.replace("\\u002F", "/").replace("\\/", "/") | |
| user_pat = re.escape(username).replace("@", "(?:@|%40)") | |
| hits = [(m.start(), m.group(1)) | |
| for m in re.finditer(rf"/photos/{user_pat}/(\d{{8,}})", html, flags=re.I)] | |
| hits += [(m.start(), m.group(1)) | |
| for m in re.finditer(r"live\.staticflickr\.com/\d+/(\d{8,})_[0-9a-f]+", html)] | |
| if not hits: | |
| hits = [(m.start(), m.group(1)) | |
| for m in re.finditer(r'"id"\s*:\s*"(\d{9,12})"', html)] | |
| return list(dict.fromkeys(i for _, i in sorted(hits))) | |
| base = f"https://www.flickr.com/photos/{username}/" | |
| ids, page = [], 1 | |
| stats["detail"] = [] | |
| while len(ids) < wanted: | |
| if page > max_pages: | |
| stats["stop"] = f"page limit ({max_pages}) reached" | |
| break | |
| variants = [base] if page == 1 else [ | |
| f"{base}page{page}/", f"{base}page{page}", f"{base}?page={page}"] | |
| new = [] | |
| for u in variants: # try URL forms until one gives photos we don't have yet | |
| try: | |
| r = session.get(u, timeout=REQUEST_TIMEOUT, allow_redirects=True) | |
| status_code, final, size = r.status_code, r.url, len(r.text) | |
| r.raise_for_status() | |
| new = [i for i in parse_ids(r.text) if i not in ids] | |
| stats["detail"].append( | |
| f"{u} → HTTP {status_code}, {size} bytes, {len(new)} new photos" | |
| + (f", redirected to {final}" if final.rstrip('/') != u.rstrip('/') else "")) | |
| except Exception as exc: | |
| stats["detail"].append(f"{u} → {type(exc).__name__}: {exc}") | |
| continue | |
| if new: | |
| break | |
| stats["pages"] += 1 | |
| stats["ids_per_page"].append(len(new)) | |
| if not new: | |
| stats["stop"] = f"page {page} gave no new photos with any URL form" | |
| break | |
| ids += new | |
| page += 1 | |
| ids = ids[:wanted] | |
| def fetch(pid): | |
| if pid == first["photo_id"]: | |
| return first | |
| err = "" | |
| for attempt in range(2): | |
| try: | |
| return flickr_photo_via_html( | |
| username, pid, f"https://www.flickr.com/photos/{username}/{pid}/") | |
| except Exception as exc: | |
| err = f"{type(exc).__name__}: {exc}" | |
| time.sleep(1 + attempt) | |
| stats["failed"] += 1 | |
| if len(stats["errors"]) < 5: | |
| stats["errors"].append(f"photo {pid}: {err}") | |
| return None | |
| # Fetch photo pages in batches; stop as soon as enough succeeded (failed ones are replaced). | |
| items = [] | |
| with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: | |
| for start in range(0, len(ids), MAX_WORKERS): | |
| if len(items) >= max_images: | |
| break | |
| items += [it for it in pool.map(fetch, ids[start:start + MAX_WORKERS]) if it] | |
| return items | |
| def flickr_photo_and_stream(url, max_images, include_stream=True, max_pages=100): | |
| """Photo page -> the photographer's public photos, newest to oldest, up to max_images.""" | |
| first = flickr_photo_candidate(url) | |
| if not first: | |
| return None | |
| if not include_stream: | |
| return [first] | |
| username, _ = extract_flickr_photo_id(url) | |
| items = flickr_stream_via_html(username, max_images, first, max_pages) | |
| items.sort(key=lambda s: s.get("date_uploaded", ""), reverse=True) | |
| return items or [first] | |
| # ============================================================ | |
| # GENERIC HTML | |
| # ============================================================ | |
| def page_publication_date(soup, html): | |
| """Try common publication-date markers on ordinary web pages.""" | |
| for attrs in [ | |
| {"property": "article:published_time"}, {"name": "article:published_time"}, | |
| {"itemprop": "datePublished"}, {"name": "date"}, {"name": "pubdate"}, | |
| {"property": "og:published_time"}, | |
| ]: | |
| tag = soup.find("meta", attrs=attrs) | |
| if tag and tag.get("content"): | |
| d = iso_or_empty(tag["content"]) | |
| if d: | |
| return d | |
| m = re.search(r'"datePublished"\s*:\s*"([^"]+)"', html) | |
| if m: | |
| d = iso_or_empty(m.group(1)) | |
| if d: | |
| return d | |
| t = soup.find("time", attrs={"datetime": True}) | |
| return iso_or_empty(t["datetime"]) if t else "" | |
| def extract_image_candidates_from_html(page_url, html): | |
| soup = BeautifulSoup(html, "html.parser") | |
| published = page_publication_date(soup, html) | |
| raw = [] | |
| for img in soup.find_all("img"): | |
| alt = img.get("alt", "") | |
| for attr in ["src", "data-src", "data-original", "data-lazy-src", "data-image", "data-url"]: | |
| if img.get(attr): | |
| raw.append((img[attr], alt)) | |
| for attr in ["srcset", "data-srcset"]: | |
| if img.get(attr): | |
| for part in img[attr].split(","): | |
| bits = part.strip().split() | |
| if bits: | |
| raw.append((bits[0], alt)) | |
| for source in soup.find_all("source"): | |
| if source.get("srcset"): | |
| for part in source["srcset"].split(","): | |
| bits = part.strip().split() | |
| if bits: | |
| raw.append((bits[0], "")) | |
| for meta in soup.find_all("meta", attrs={"property": "og:image"}): | |
| if meta.get("content"): | |
| raw.append((meta["content"], "og:image")) | |
| out = [] | |
| for value, alt in raw: | |
| value = value.strip() | |
| if not value or value.startswith("data:"): | |
| continue | |
| absolute = urljoin(page_url, value) | |
| if valid_http_url(absolute): | |
| out.append(make_item( | |
| image_url=absolute, page_url=page_url, source="HTML", | |
| alt=alt, date_uploaded=published, | |
| )) | |
| return out | |
| def generic_page_images(page_url, max_images=50): | |
| r = session.get(page_url, timeout=REQUEST_TIMEOUT, allow_redirects=True) | |
| r.raise_for_status() | |
| ctype = r.headers.get("Content-Type", "").lower() | |
| if "html" not in ctype: | |
| if ctype.startswith("image/"): | |
| return [make_item(image_url=r.url, page_url=page_url, source="Direct image")] | |
| raise RuntimeError(f"The page did not return HTML. Content-Type: {ctype}") | |
| seen, unique = set(), [] | |
| for item in extract_image_candidates_from_html(r.url, r.text): | |
| key = item["image_url"].split("#")[0] | |
| if key in seen: | |
| continue | |
| seen.add(key) | |
| unique.append(item) | |
| if len(unique) >= max_images: | |
| break | |
| return unique | |
| # ============================================================ | |
| # DOWNLOAD + METADATA | |
| # ============================================================ | |
| def download_image(item, output_dir, index, min_width=0, min_height=0): | |
| image_url = item["image_url"] | |
| try: | |
| r = session.get(image_url, timeout=IMAGE_TIMEOUT, stream=True, allow_redirects=True) | |
| r.raise_for_status() | |
| ctype = r.headers.get("Content-Type", "").lower() | |
| if ctype and not ctype.startswith("image/") and "octet-stream" not in ctype: | |
| return None, f"Not an image: {ctype}" | |
| data = r.content | |
| if not data: | |
| return None, "Empty response" | |
| try: | |
| image = Image.open(io.BytesIO(data)) | |
| width, height = image.size | |
| image_format = (image.format or "JPEG").lower() | |
| except Exception as exc: | |
| return None, f"Invalid image: {exc}" | |
| if min_width and width < min_width: | |
| return None, f"Too narrow: {width}px" | |
| if min_height and height < min_height: | |
| return None, f"Too short: {height}px" | |
| exif_md = extract_exif_metadata(image) | |
| ext = FORMAT_EXT.get(image_format) or CONTENT_TYPE_EXT.get(ctype.split(";")[0], ".jpg") | |
| stem = Path(urlparse(image_url).path).stem | |
| stem = safe_filename(stem, f"image_{index:05d}") if stem else f"image_{index:05d}" | |
| filename = safe_filename(f"{index:05d}_{stem}{ext}", f"image_{index:05d}{ext}") | |
| output_path = Path(output_dir) / filename | |
| output_path.write_bytes(data) | |
| # ---- Date selection: publication date first ---- | |
| published = item.get("date_uploaded", "") | |
| taken = item.get("date_taken", "") | |
| date_used, date_source = "", "" | |
| for value, label in [ | |
| (published, "Publication date"), | |
| (taken, "Flickr date taken"), | |
| (exif_md["exif_datetime_original"], "EXIF DateTimeOriginal"), | |
| (exif_md["exif_datetime"], "EXIF DateTime"), | |
| ]: | |
| if value: | |
| date_used, date_source = value, label | |
| break | |
| parsed = parse_date(date_used) | |
| ok = not pd.isna(parsed) | |
| record = dict(item) | |
| record.update({ | |
| "filename": filename, | |
| "local_file": str(output_path), | |
| "final_url": r.url, | |
| "width": width, | |
| "height": height, | |
| "aspect_ratio": round(width / height, 3) if height else "", | |
| "format": image_format, | |
| "file_size_bytes": len(data), | |
| "file_size_mb": round(len(data) / 1024 / 1024, 3), | |
| "sha256": hashlib.sha256(data).hexdigest(), | |
| "date_published": iso_or_empty(published), | |
| "date_used": date_used, | |
| "date_source": date_source, | |
| "date_parsed": parsed.strftime("%Y-%m-%d") if ok else "", | |
| "week": parsed.strftime("%G-W%V") if ok else "", | |
| "download_status": "OK", | |
| }) | |
| record.update(exif_md) | |
| return record, None | |
| except Exception as exc: | |
| return None, str(exc) | |
| # ============================================================ | |
| # DATAFRAMES, PLOT, EXPORTS | |
| # ============================================================ | |
| PREFERRED_COLUMNS = [ | |
| "filename", "title", "owner", "photo_id", "source", | |
| "page_url", "image_url", "final_url", | |
| "date_published", "date_used", "date_source", "date_parsed", "week", | |
| "date_taken", "date_uploaded", | |
| "width", "height", "aspect_ratio", "format", "file_size_bytes", "file_size_mb", | |
| "license", "tags", "description", "views", "alt", | |
| "exif_make", "exif_model", "exif_software", | |
| "exif_datetime", "exif_datetime_original", "exif_datetime_digitized", | |
| "exif_orientation", "exif_lens_model", "exif_focal_length", "exif_iso", | |
| "exif_exposure_time", "exif_f_number", | |
| "exif_gps_latitude", "exif_gps_longitude", "exif_gps_altitude", | |
| "sha256", "download_status", | |
| ] | |
| def records_to_dataframe(records): | |
| if not records: | |
| return pd.DataFrame() | |
| df = pd.DataFrame(records) | |
| existing = [c for c in PREFERRED_COLUMNS if c in df.columns] | |
| rest = [c for c in df.columns if c not in existing and c != "local_file"] | |
| return df[existing + rest] | |
| def count_nonempty(df, col): | |
| if df.empty or col not in df.columns: | |
| return 0 | |
| return int(df[col].fillna("").astype(str).str.len().gt(0).sum()) | |
| def create_weekly_dataframe(df): | |
| cols = ["week", "week_start", "number_of_pictures"] | |
| if df.empty: | |
| return pd.DataFrame(columns=cols) | |
| temp = df.copy() | |
| temp["_date"] = pd.to_datetime(temp["date_parsed"], errors="coerce") | |
| temp = temp[temp["_date"].notna()].copy() | |
| if temp.empty: | |
| return pd.DataFrame(columns=cols) | |
| temp["week_start"] = (temp["_date"] - pd.to_timedelta(temp["_date"].dt.weekday, unit="D")).dt.normalize() | |
| counts = temp.groupby("week_start").size() | |
| # Continuous timeline: weeks without pictures appear as 0 | |
| full = pd.date_range(counts.index.min(), counts.index.max(), freq="7D") | |
| counts = counts.reindex(full, fill_value=0) | |
| weekly = pd.DataFrame({ | |
| "week": [d.strftime("%G-W%V") for d in full], | |
| "week_start": [d.strftime("%Y-%m-%d") for d in full], | |
| "number_of_pictures": counts.values.astype(int), | |
| }) | |
| return weekly | |
| def create_weekly_plot(weekly_df, output_path): | |
| if weekly_df.empty: | |
| return None | |
| from matplotlib.figure import Figure # no pyplot: safe inside Gradio worker threads | |
| n = len(weekly_df) | |
| fig = Figure(figsize=(max(10, min(24, n * 0.4)), 6)) | |
| ax = fig.subplots() | |
| x = np.arange(n) | |
| y = weekly_df["number_of_pictures"].astype(int).values | |
| bars = ax.bar(x, y, color="#4C78A8") | |
| labels = weekly_df["week"].tolist() | |
| step = max(1, math.ceil(n / 30)) | |
| ax.set_xticks(x[::step]) | |
| ax.set_xticklabels(labels[::step], rotation=45, ha="right") | |
| if n <= 40: | |
| for b, v in zip(bars, y): | |
| if v: | |
| ax.annotate(str(v), (b.get_x() + b.get_width() / 2, v), | |
| ha="center", va="bottom", fontsize=8) | |
| ax.set_xlabel("Week (Monday start)") | |
| ax.set_ylabel("Number of pictures") | |
| ax.set_title("Number of pictures per week") | |
| ax.grid(axis="y", alpha=0.25) | |
| ax.set_ylim(0, max(1, y.max()) * 1.15) | |
| fig.tight_layout() | |
| fig.savefig(output_path, dpi=150) | |
| return str(output_path) | |
| def create_xlsx(df, weekly_df, output_path): | |
| summary_df = pd.DataFrame([ | |
| {"metric": "Number of pictures", "value": len(df)}, | |
| {"metric": "Pictures with publication date", "value": count_nonempty(df, "date_published")}, | |
| {"metric": "Pictures with any usable date", "value": count_nonempty(df, "date_parsed")}, | |
| {"metric": "Pictures with camera EXIF", "value": count_nonempty(df, "exif_model")}, | |
| {"metric": "Pictures with GPS", "value": count_nonempty(df, "exif_gps_latitude")}, | |
| ]) | |
| with pd.ExcelWriter(output_path, engine="openpyxl") as writer: | |
| df.to_excel(writer, sheet_name="Metadata", index=False) | |
| weekly_df.to_excel(writer, sheet_name="Weekly_counts", index=False) | |
| summary_df.to_excel(writer, sheet_name="Summary", index=False) | |
| for name in ["Metadata", "Weekly_counts", "Summary"]: | |
| ws = writer.book[name] | |
| ws.freeze_panes = "A2" | |
| for col in ws.columns: | |
| longest = max((len(str(c.value)) for c in col if c.value is not None), default=0) | |
| ws.column_dimensions[col[0].column_letter].width = min(max(longest + 2, 10), 45) | |
| def build_zip(records, zip_path): | |
| """ZIP with the pictures only (metadata/plot are separate downloads).""" | |
| with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as z: | |
| for rec in records: | |
| lf = rec.get("local_file") | |
| if lf and os.path.exists(lf): | |
| z.write(lf, arcname=rec.get("filename", Path(lf).name)) | |
| # ============================================================ | |
| # MAIN | |
| # ============================================================ | |
| def empty_result(message): | |
| return [], message, [], pd.DataFrame(), pd.DataFrame(), None, None, None, None | |
| def find_images(url, max_images, min_width, min_height, remove_duplicates, | |
| include_stream=True, max_pages=3): | |
| url = clean_url(url) | |
| if not url: | |
| return empty_result("❌ Please enter a URL.") | |
| if not valid_http_url(url): | |
| return empty_result("❌ Invalid HTTP/HTTPS URL.") | |
| max_images = min(int(max_images), MAX_IMAGES_HARD_LIMIT) | |
| max_pages = max(1, int(max_pages or 1)) | |
| min_width = int(min_width or 0) | |
| min_height = int(min_height or 0) | |
| STREAM_STATS.clear() | |
| job_dir = BASE_DIR / f"{int(time.time())}_{make_hash(url)}" | |
| job_dir.mkdir(parents=True, exist_ok=True) | |
| # ---- Find candidates ---- | |
| try: | |
| photo_list = ( | |
| flickr_photo_and_stream(url, max_images, include_stream, max_pages) | |
| if is_flickr_url(url) else None | |
| ) | |
| if photo_list: | |
| mode_message = "Flickr photo page + photostream (HTML mode, one request per photo)" | |
| candidates = photo_list | |
| else: | |
| mode_message = "HTML webpage mode" | |
| candidates = generic_page_images(url, max_images) | |
| except Exception as exc: | |
| return empty_result(f"❌ Could not retrieve the page.\n\n{type(exc).__name__}: {exc}") | |
| if not candidates: | |
| return empty_result("⚠️ No image candidates were found.") | |
| # ---- URL deduplication ---- | |
| if remove_duplicates: | |
| seen, unique = set(), [] | |
| for item in candidates: | |
| key = item.get("image_url", "").split("#")[0] | |
| if key not in seen: | |
| seen.add(key) | |
| unique.append(item) | |
| candidates = unique | |
| # ---- Download ---- | |
| candidates = candidates[: max_images + max(5, max_images // 5)] | |
| downloaded, errors = [], [] | |
| def work(args): | |
| index, item = args | |
| return item, download_image(item, job_dir, index, min_width, min_height) | |
| # Batches: stop as soon as we have max_images, and failed ones get replaced by spares. | |
| with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: | |
| for start in range(0, len(candidates), MAX_WORKERS): | |
| if len(downloaded) >= max_images: | |
| break | |
| batch = list(enumerate(candidates[start:start + MAX_WORKERS], start=start + 1)) | |
| for item, (result, error) in pool.map(work, batch): | |
| if result: | |
| downloaded.append(result) | |
| else: | |
| errors.append({"url": item.get("image_url", ""), "error": error}) | |
| downloaded = downloaded[:max_images] | |
| # ---- Exact duplicate detection ---- | |
| if remove_duplicates and downloaded: | |
| seen_hashes, unique_dl = set(), [] | |
| for rec in downloaded: | |
| if rec["sha256"] in seen_hashes: | |
| try: | |
| os.remove(rec["local_file"]) | |
| except Exception: | |
| pass | |
| continue | |
| seen_hashes.add(rec["sha256"]) | |
| unique_dl.append(rec) | |
| downloaded = unique_dl | |
| # ---- Tables & files ---- | |
| df = records_to_dataframe(downloaded) | |
| weekly_df = create_weekly_dataframe(df) | |
| csv_path = job_dir / "metadata.csv" | |
| xlsx_path = job_dir / "metadata.xlsx" | |
| plot_path = job_dir / "weekly_picture_count.png" | |
| df.to_csv(csv_path, index=False, encoding="utf-8-sig") | |
| create_xlsx(df, weekly_df, xlsx_path) | |
| plot_error = "" | |
| try: | |
| create_weekly_plot(weekly_df, plot_path) | |
| except Exception as exc: | |
| plot_error = f"{type(exc).__name__}: {exc}" | |
| gallery_items = [ | |
| (r["local_file"], r.get("title") or r.get("filename")) | |
| for r in downloaded if r.get("local_file") | |
| ] | |
| lines = [ | |
| f"### {mode_message}", "", | |
| f"**Candidates found:** {len(candidates)}", | |
| f"**Images downloaded:** {len(downloaded)}", | |
| f"**Failed downloads:** {len(errors)}", "", | |
| f"**Images with publication date:** {count_nonempty(df, 'date_published')}", | |
| f"**Images with any usable date:** {count_nonempty(df, 'date_parsed')}", | |
| f"**Images with camera EXIF:** {count_nonempty(df, 'exif_model')}", | |
| f"**Images with GPS EXIF:** {count_nonempty(df, 'exif_gps_latitude')}", "", | |
| f"**Weeks represented:** {len(weekly_df)}", | |
| ] | |
| if STREAM_STATS.get("pages") is not None and STREAM_STATS: | |
| st = STREAM_STATS | |
| lines += ["", "### Photostream diagnostics", | |
| f"- Pages scraped: {st['pages']} (new photos per page: {st['ids_per_page']})", | |
| f"- Photo pages that failed to load: {st['failed']}"] | |
| if st.get("stop"): | |
| lines.append(f"- Stopped because: {st['stop']}") | |
| lines += [f"- {e}" for e in st["errors"][:3]] | |
| lines += ["", "Requests made:"] + [f" - {d}" for d in st.get("detail", [])[:12]] | |
| if len(downloaded) < max_images and photo_list: | |
| lines += ["", f"⚠️ Got {len(downloaded)} of {max_images} requested. See diagnostics above."] | |
| if plot_error: | |
| lines += ["", f"⚠️ Weekly plot failed: {plot_error}"] | |
| elif weekly_df.empty and downloaded: | |
| lines += ["", "⚠️ No usable dates, so no weekly plot."] | |
| if downloaded and count_nonempty(df, "date_published") == 0: | |
| lines += ["", "⚠️ No publication date was found."] | |
| if errors: | |
| lines += ["", "### Download errors"] + [f"- {e['error']}" for e in errors[:10]] | |
| return ( | |
| gallery_items, | |
| "\n".join(lines), | |
| downloaded, | |
| df, | |
| weekly_df, | |
| str(plot_path) if plot_path.exists() else None, | |
| str(xlsx_path), | |
| str(csv_path), | |
| str(plot_path) if plot_path.exists() else None, | |
| ) | |
| def create_zip(items): | |
| if not items: | |
| return None, "❌ There are no downloaded images." | |
| try: | |
| job_dir = Path(items[0]["local_file"]).parent | |
| zip_path = job_dir.parent / f"images_{time.strftime('%Y%m%d_%H%M%S')}.zip" | |
| build_zip(items, zip_path) | |
| size_mb = zip_path.stat().st_size / 1024 / 1024 | |
| return str(zip_path), ( | |
| f"### ✅ ZIP created\n\nImages: {len(items)}\n\nZIP size: {size_mb:.2f} MB\n\n" | |
| "The ZIP contains the pictures only. Download the metadata and plot files separately above." | |
| ) | |
| except Exception as exc: | |
| return None, f"❌ ZIP creation failed: {exc}" | |
| def clear_all(): | |
| # url, gallery, status, image_state, metadata, weekly_table, weekly_plot, xlsx, csv, plot_file, zip, zip_status | |
| return "", [], "Ready.", [], pd.DataFrame(), pd.DataFrame(), None, None, None, None, None, "" | |
| # ============================================================ | |
| # UI | |
| # ============================================================ | |
| DESCRIPTION = """ | |
| ## 🖼️ Web & Flickr Image Downloader | |
| Download images from Flickr or ordinary webpages and extract metadata. | |
| ### Supported URLs | |
| - **Flickr photo page:** `https://www.flickr.com/photos/<user>/with/<photo_id>` | |
| - Any ordinary webpage or direct image URL | |
| No Flickr API is used: everything is read from the public web pages. | |
| ### Date used for the weekly analysis (first available) | |
| 1. **Publication date** (Flickr upload date, or the page's published date on ordinary sites) | |
| 2. Flickr date taken | |
| 3. EXIF DateTimeOriginal | |
| 4. EXIF DateTime | |
| Weeks start on Monday (ISO week numbering). | |
| ### Exports | |
| Separate downloads: `metadata.xlsx`, `metadata.csv` and the weekly bar chart (PNG). The ZIP contains the pictures only. | |
| """ | |
| with gr.Blocks(title=APP_TITLE) as demo: | |
| gr.Markdown("# 🖼️ Web & Flickr Image Downloader") | |
| gr.Markdown(DESCRIPTION) | |
| with gr.Row(): | |
| with gr.Column(scale=4): | |
| url_input = gr.Textbox( | |
| label="Webpage or Flickr URL", | |
| placeholder="https://www.flickr.com/photos/snapnpiks0304/with/55555288566", | |
| lines=2, | |
| ) | |
| with gr.Column(scale=1): | |
| max_images = gr.Slider(1, 500, value=50, step=1, label="Maximum images") | |
| with gr.Row(): | |
| min_width = gr.Number(value=0, minimum=0, maximum=10000, step=100, label="Minimum width (px)") | |
| min_height = gr.Number(value=0, minimum=0, maximum=10000, step=100, label="Minimum height (px)") | |
| remove_duplicates = gr.Checkbox(value=True, label="Remove duplicates") | |
| include_stream = gr.Checkbox( | |
| value=True, | |
| label="Photo page: also download the photographer's other photos", | |
| ) | |
| max_pages = gr.Number( | |
| value=3, minimum=1, maximum=100, step=1, precision=0, | |
| label="Photostream pages to scrape (stops earlier if Maximum images is reached)", | |
| ) | |
| with gr.Row(): | |
| find_button = gr.Button("🔎 Find & Analyze Images", variant="primary") | |
| clear_button = gr.Button("🧹 Clear") | |
| status = gr.Markdown("Ready.") | |
| gallery = gr.Gallery( | |
| label="Downloaded images", columns=5, rows=3, | |
| height="auto", object_fit="contain", allow_preview=True, | |
| ) | |
| image_state = gr.State([]) | |
| gr.Markdown("## 📋 Image metadata") | |
| metadata_table = gr.Dataframe(value=pd.DataFrame(), interactive=False, wrap=True, label="Metadata") | |
| gr.Markdown("## 📅 Pictures per week") | |
| weekly_table = gr.Dataframe(value=pd.DataFrame(), interactive=False, label="Weekly counts") | |
| weekly_plot = gr.Image(label="Number of pictures per week", type="filepath") | |
| gr.Markdown("## 📥 Downloads") | |
| with gr.Row(): | |
| xlsx_file = gr.File(label="Excel metadata") | |
| csv_file = gr.File(label="CSV metadata") | |
| plot_file = gr.File(label="Weekly plot (PNG)") | |
| gr.Markdown("## 📦 Pictures ZIP") | |
| with gr.Row(): | |
| zip_button = gr.Button("📦 Create Pictures ZIP", variant="primary") | |
| zip_file = gr.File(label="Pictures ZIP") | |
| zip_status = gr.Markdown("") | |
| find_button.click( | |
| fn=find_images, | |
| inputs=[url_input, max_images, min_width, min_height, remove_duplicates, | |
| include_stream, max_pages], | |
| outputs=[gallery, status, image_state, metadata_table, weekly_table, | |
| weekly_plot, xlsx_file, csv_file, plot_file], | |
| ) | |
| # The ZIP button reads the generated files straight from the file components, | |
| # so no extra State objects (and no race condition) are needed. | |
| zip_button.click( | |
| fn=create_zip, | |
| inputs=[image_state], | |
| outputs=[zip_file, zip_status], | |
| ) | |
| clear_button.click( | |
| fn=clear_all, | |
| inputs=[], | |
| outputs=[url_input, gallery, status, image_state, metadata_table, | |
| weekly_table, weekly_plot, xlsx_file, csv_file, plot_file, zip_file, zip_status], | |
| ) | |
| if __name__ == "__main__": | |
| demo.launch() |