DEV Community: Cizze R

3 API Checks I Run Before a Website Launch

Cizze R — Wed, 12 Aug 2026 12:00:01 +0000

3 API Checks I Run Before a Website Launch

A website can return 200 OK and still launch with the wrong certificate, a staging canonical URL, or a mobile layout nobody checked. I use three small API checks to create a timestamped launch evidence pack before DNS cutover or a major release.

The checks cover infrastructure with Domain Intelligence Suite, document metadata with Link Preview & Metadata Extractor, and rendered output with Website Screenshot API. Each tool answers a different question, which keeps the Python orchestration simple and the failures specific.

Use one boring Actor client

All three Actors can return a dataset item through Apify's synchronous endpoint. A shared client handles authentication, timeouts, and response validation without hiding which Actor failed.

import os
from typing import Any

import requests

APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
API_ROOT = "https://api.apify.com/v2/acts"


def run_actor(actor_id: str, payload: dict[str, Any]) -> dict[str, Any]:
    url = f"{API_ROOT}/{actor_id}/run-sync-get-dataset-items"
    response = requests.post(
        url,
        params={"token": APIFY_TOKEN},
        json=payload,
        timeout=120,
    )
    response.raise_for_status()

    items = response.json()
    if not isinstance(items, list) or len(items) != 1:
        raise RuntimeError(
            f"{actor_id} returned {len(items) if isinstance(items, list) else 'invalid'} items"
        )
    return items[0]

In CI, set APIFY_TOKEN through the platform's secret store. The fallback placeholder makes the example copyable without encouraging tokens in source control.

Check 1: DNS and TLS match the launch plan

Before checking page content, confirm that the domain resolves and presents a healthy certificate. The domain Actor runs DNS and SSL modules independently, so the code checks each module for its own error.

DOMAIN_ACTOR = "weeknds~domain-intelligence-suite"


def check_domain(domain: str, minimum_tls_days: int = 14) -> dict:
    result = run_actor(
        DOMAIN_ACTOR,
        {
            "domain": domain,
            "modules": ["dns", "ssl"],
            "dnsRecordTypes": ["A", "AAAA", "CNAME", "NS"],
            "sslPort": 443,
        },
    )

    findings = []
    dns = result.get("dns", {})
    ssl = result.get("ssl", {})

    if dns.get("error"):
        findings.append(f"DNS lookup failed: {dns['error']}")
    else:
        records = dns.get("records", {})
        if not any(records.get(kind) for kind in ("A", "AAAA", "CNAME")):
            findings.append("No A, AAAA, or CNAME record found")

    if ssl.get("error"):
        findings.append(f"TLS check failed: {ssl['error']}")
    else:
        certificate = ssl.get("certificate", {})
        if certificate.get("expired"):
            findings.append("TLS certificate is expired")
        days = certificate.get("days_remaining")
        if isinstance(days, int) and days < minimum_tls_days:
            findings.append(f"TLS certificate expires in {days} days")

    return {
        "passed": not findings,
        "findings": findings,
        "evidence": result,
    }

This is a launch gate, not a full security audit. It verifies the routing and certificate facts most likely to be broken during cutover. Keep expected IP addresses or nameservers in release configuration if the job must also prove that DNS points to a particular provider.

Check 2: Validate what crawlers will read

A rendered page can look correct while its canonical, description, or social image still references staging. The metadata Actor fetches the deployed URL over HTTP and returns document metadata plus response details.

The response can expose fields as nested data or literal Open Graph keys, so a small recursive search keeps the policy independent from presentation.

from collections.abc import Mapping
from urllib.parse import urlparse

METADATA_ACTOR = "weeknds~link-preview-metadata-extractor"


def find_value(value: Mapping, *wanted: str):
    for key, child in value.items():
        if str(key) in wanted and child not in (None, "", []):
            return child
        if isinstance(child, Mapping):
            match = find_value(child, *wanted)
            if match not in (None, "", []):
                return match
    return None


def check_metadata(url: str, expected_host: str) -> dict:
    result = run_actor(
        METADATA_ACTOR,
        {
            "url": url,
            "includeFavicon": True,
            "userAgent": "Launch-Evidence-Check/1.0",
            "timeout": 20,
        },
    )
    findings = []

    if result.get("error"):
        findings.append(result["error"])
    else:
        status = find_value(result, "status_code")
        title = find_value(result, "og:title", "title")
        canonical = find_value(result, "canonical_url", "canonical", "og:url")
        social_image = find_value(result, "og:image")
        robots = str(find_value(result, "robots") or "").lower()

        if status and not 200 <= int(status) < 300:
            findings.append(f"Metadata request returned HTTP {status}")
        if not title:
            findings.append("No document or Open Graph title found")
        if not canonical:
            findings.append("Canonical URL is missing")
        elif urlparse(str(canonical)).hostname != expected_host:
            findings.append(f"Canonical URL does not use {expected_host}")
        if not social_image or urlparse(str(social_image)).scheme != "https":
            findings.append("Open Graph image must be an absolute HTTPS URL")
        if "noindex" in robots:
            findings.append("Robots metadata contains noindex")

    return {
        "passed": not findings,
        "findings": findings,
        "evidence": result,
    }

The title and social image requirements are product policy, not web standards. Adjust the rules for a private dashboard, documentation site, or marketing page, but keep the expected canonical host explicit.

Check 3: Capture the layouts humans will see

Metadata checks cannot catch a cookie banner covering the call to action or a navigation menu overflowing on mobile. I capture desktop light mode and mobile dark mode as a compact visual pair.

from pathlib import Path

SCREENSHOT_ACTOR = "weeknds~website-screenshot-api"


def capture_view(url: str, viewport: str, dark_mode: bool) -> dict:
    result = run_actor(
        SCREENSHOT_ACTOR,
        {
            "url": url,
            "viewport": viewport,
            "fullPage": True,
            "delay": 1500,
            "format": "png",
            "darkMode": dark_mode,
        },
    )
    if not result.get("success") or not result.get("screenshotUrl"):
        raise RuntimeError(result.get("error") or "Screenshot capture failed")
    return result


def download_screenshot(result: dict, destination: Path) -> Path:
    response = requests.get(result["screenshotUrl"], timeout=60)
    response.raise_for_status()
    if not response.headers.get("content-type", "").startswith("image/"):
        raise RuntimeError("Screenshot URL did not return an image")

    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_suffix(destination.suffix + ".tmp")
    temporary.write_bytes(response.content)
    temporary.replace(destination)
    return destination

A fixed viewport and delay make repeated captures easier to compare. Dynamic ads, clocks, personalization, and animations can still change between runs, so visual evidence needs human review rather than a byte-for-byte equality gate.

Assemble a timestamped evidence pack

The orchestrator runs all checks, saves the complete structured responses, and downloads screenshots next to the JSON. It exits non-zero when an automated gate fails while still preserving evidence from completed checks.

import json
from datetime import datetime, timezone
from pathlib import Path


def build_launch_pack(url: str, domain: str, output_root: Path) -> Path:
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    pack = output_root / f"{timestamp}-{domain}"
    pack.mkdir(parents=True, exist_ok=False)

    checks = {
        "domain": check_domain(domain),
        "metadata": check_metadata(url, domain),
    }

    screenshot_specs = [
        ("desktop-light", "desktop", False),
        ("mobile-dark", "mobile", True),
    ]
    screenshots = {}
    for name, viewport, dark_mode in screenshot_specs:
        result = capture_view(url, viewport, dark_mode)
        path = download_screenshot(result, pack / f"{name}.png")
        screenshots[name] = {"file": path.name, "actor_result": result}

    report = {
        "url": url,
        "domain": domain,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "checks": checks,
        "screenshots": screenshots,
    }
    (pack / "report.json").write_text(
        json.dumps(report, indent=2, sort_keys=True),
        encoding="utf-8",
    )

    failed = [name for name, result in checks.items() if not result["passed"]]
    if failed:
        raise RuntimeError("Launch checks failed: " + ", ".join(failed))
    return pack


if __name__ == "__main__":
    evidence = build_launch_pack(
        "https://www.example.com/",
        "www.example.com",
        Path("launch-evidence"),
    )
    print(evidence)

Run this against the exact public hostname users will visit. If DNS is still private before cutover, run it from an authorized network or use a temporary validation hostname and repeat the check after the public change.

Pricing and release timing

Current Store pricing starts at $5 per 1,000 domain intelligence results for Domain Intelligence Suite, $2 per 1,000 extractions for Link Preview & Metadata Extractor, and $3 per 1,000 captures for Website Screenshot API. One pass with one domain check, one metadata extraction, and two screenshots is $0.013 in Actor charges, so 100 passes are $1.30. Apify platform usage may also apply; confirm live pricing before a high-volume rollout.

Do not run the pack only before deployment. Run it once after the public DNS or CDN change has propagated, because that is when resolver paths, production headers, certificates, and canonical URLs can differ from staging.

Store the evidence pack with the release ID and a short retention period. Review both images at their native size, verify any flagged field against the raw Actor response, and record the approver separately so a screenshot folder is never mistaken for an approval decision.

How to Build an Attack-Surface Inventory from Certificate Transparency

Cizze R — Mon, 10 Aug 2026 12:00:50 +0000

How to Build an Attack-Surface Inventory from Certificate Transparency

A domain inventory maintained in a spreadsheet starts drifting as soon as another team creates a staging host or replaces a certificate. I built a small Python job that discovers names from Certificate Transparency, records what the domain exposes today, and reports only the changes worth reviewing.

The Domain Intelligence Suite provides the WHOIS, DNS, TLS, and certificate-derived subdomain data. Python turns that response into a durable inventory rather than another JSON file nobody checks.

Treat Certificate Transparency as discovery, not proof

Public certificate authorities submit certificates to Certificate Transparency logs. Searching those logs can reveal names such as api.example.com, staging.example.com, and wildcard entries that were included in issued certificates.

That makes CT useful for discovery, but it is not a perfect inventory. A name in a certificate might be retired, internal, misspelled, or no longer resolve. A service using a private certificate will not appear. The right workflow is:

Discover candidate names from CT.
Normalize and deduplicate them.
Resolve or verify candidates you intend to monitor.
Compare the latest set with the previous snapshot.
Ask an owner to classify unexpected additions.

Do not turn a new certificate name directly into an incident. Treat it as a lead that needs context.

Fetch the domain report in one request

Apify's synchronous dataset endpoint waits for the Actor and returns its dataset items in the response. This works well for a scheduled inventory job because there is no run ID or polling loop to maintain.

import os
from typing import Any

import requests

APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~domain-intelligence-suite"
RUN_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def fetch_domain_report(domain: str) -> dict[str, Any]:
    response = requests.post(
        RUN_URL,
        params={"token": APIFY_TOKEN},
        json={
            "domain": domain,
            "modules": ["dns", "ssl", "subdomains"],
            "dnsRecordTypes": ["A", "AAAA", "CNAME", "MX", "NS"],
            "maxSubdomains": 1000,
            "includeWildcardSubdomains": True,
        },
        timeout=120,
    )
    response.raise_for_status()

    items = response.json()
    if not isinstance(items, list) or len(items) != 1:
        raise RuntimeError(f"Expected one domain result, got {len(items)}")
    return items[0]

Keep APIFY_TOKEN in a secret store for cron or CI. YOUR_APIFY_TOKEN shows the required value without placing a real credential in the script.

The Actor runs its modules independently. One module can contain an error while others still return useful data, so check module results instead of treating any partial failure as an empty inventory.

Normalize the names before storing them

CT data often includes wildcard names, duplicates with different casing, and entries that do not belong under the requested suffix after loose matching. Normalize the hostname and apply a strict domain boundary.

import re

HOSTNAME = re.compile(
    r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)*"
    r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$"
)


def normalize_hostname(value: str, root_domain: str) -> str | None:
    name = value.strip().lower().rstrip(".")
    if name.startswith("*."):
        name = name[2:]

    root = root_domain.lower().rstrip(".")
    if name != root and not name.endswith(f".{root}"):
        return None
    if not HOSTNAME.fullmatch(name):
        return None
    return name


def discovered_names(report: dict, root_domain: str) -> set[str]:
    module = report.get("subdomains", {})
    if module.get("error"):
        raise RuntimeError(f"CT discovery failed: {module['error']}")

    candidates = module.get("subdomains", [])
    normalized = {
        name
        for candidate in candidates
        if (name := normalize_hostname(str(candidate), root_domain))
    }
    normalized.add(root_domain.lower().rstrip("."))
    return normalized

Removing *. does not mean every possible wildcard hostname exists. It records the certificate scope as the base name, which is enough to flag the wildcard for review without inventing thousands of hosts.

Save immutable snapshots in SQLite

An inventory becomes useful when it can answer “what changed?” SQLite is enough for a domain portfolio and keeps each scan as an immutable snapshot.

import json
import sqlite3
from datetime import datetime, timezone


def open_inventory(path: str = "attack-surface.sqlite3") -> sqlite3.Connection:
    connection = sqlite3.connect(path)
    connection.execute("""
        CREATE TABLE IF NOT EXISTS scans (
            id INTEGER PRIMARY KEY,
            root_domain TEXT NOT NULL,
            scanned_at TEXT NOT NULL,
            names_json TEXT NOT NULL,
            raw_report_json TEXT NOT NULL
        )
    """)
    return connection


def save_scan(connection: sqlite3.Connection, domain: str, report: dict) -> int:
    names = sorted(discovered_names(report, domain))
    cursor = connection.execute(
        """
        INSERT INTO scans (root_domain, scanned_at, names_json, raw_report_json)
        VALUES (?, ?, ?, ?)
        """,
        (
            domain,
            datetime.now(timezone.utc).isoformat(),
            json.dumps(names),
            json.dumps(report, sort_keys=True),
        ),
    )
    connection.commit()
    return int(cursor.lastrowid)

Saving the raw response matters. If a parser changes later, the original evidence still shows what DNS, TLS, and CT modules returned during that scan. Apply a retention policy if the portfolio is large rather than silently overwriting history.

Compare the two latest successful snapshots

A daily list of every hostname creates noise. The useful output is additions and removals between two successful scans.

def latest_name_sets(
    connection: sqlite3.Connection,
    domain: str,
) -> tuple[set[str], set[str]] | None:
    rows = connection.execute(
        """
        SELECT names_json
        FROM scans
        WHERE root_domain = ?
        ORDER BY scanned_at DESC
        LIMIT 2
        """,
        (domain,),
    ).fetchall()

    if len(rows) < 2:
        return None
    current = set(json.loads(rows[0][0]))
    previous = set(json.loads(rows[1][0]))
    return current, previous


def inventory_changes(connection: sqlite3.Connection, domain: str) -> dict:
    snapshots = latest_name_sets(connection, domain)
    if snapshots is None:
        return {"added": [], "removed": [], "baseline_created": True}

    current, previous = snapshots
    return {
        "added": sorted(current - previous),
        "removed": sorted(previous - current),
        "baseline_created": False,
    }

Only save a scan when the subdomain module succeeded. Saving an empty list after a timeout would make every known name appear removed and every name reappear on the next run.

Add context before paging anyone

Not every new hostname deserves the same response. A name like staging.example.com might be expected, while vpn-old.example.com could expose a forgotten service. Start with deterministic labels rather than a mysterious risk score.

SENSITIVE_LABELS = {
    "admin", "auth", "billing", "internal", "jenkins",
    "old", "staging", "test", "vpn",
}


def classify_name(hostname: str, root_domain: str) -> dict:
    relative = hostname.removesuffix(f".{root_domain}")
    labels = set(relative.split("."))
    matched = sorted(labels & SENSITIVE_LABELS)
    return {
        "hostname": hostname,
        "review": bool(matched),
        "matched_labels": matched,
    }

A label match means “review this first,” not “this system is vulnerable.” Confirm DNS resolution, ownership, intended exposure, authentication, and patch status through authorized processes. Do not scan ports or attempt access unless that activity is explicitly approved.

Cost and operating limits

The Domain Intelligence Suite is currently listed from $5 per 1,000 domain intelligence results, or $0.005 per domain result in Actor charges. Apify platform usage may also apply, and the live Actor page should be checked before increasing the schedule or domain count.

CT sources can be delayed or rate-limited, and certificate names describe issuance rather than current reachability. Run the inventory daily or weekly, retry transient module errors once with backoff, and keep the previous successful snapshot when discovery fails.

For each added name, record an owner, environment, expected exposure, and review date in the system your team already uses. The SQLite scan remains evidence of discovery; the ownership record explains whether the hostname is expected and who must remove it when it is no longer needed.

I Turned Nginx Access Logs into an IP and ASN Traffic Report

Cizze R — Fri, 31 Jul 2026 12:00:56 +0000

I Turned Nginx Access Logs into an IP and ASN Traffic Report

Raw Nginx logs told me which endpoints were busy, but not whether a traffic spike came from customers, a cloud provider, or one unexpected network. I built a small enrichment job that turns unique public IPs into a country and ASN report without putting geolocation calls in the request path.

The IP Geolocation Lookup does the lookup, and the rest is a local Python pipeline with parsing, deduplication, caching, and deliberately boring CSV output.

Start with offline enrichment

Calling a geolocation API from Nginx middleware adds latency and creates a new failure mode for every request. Access logs already contain what the report needs, so the job can run after rotation and fail independently from the application.

Use a log format you control. This example expects the conventional combined format, where the client IP is the first field:

203.0.113.42 - - [24/Jul/2026:13:04:11 +0000] "GET /api/search HTTP/1.1" 200 821 "-" "Mozilla/5.0"

The parser only extracts valid public addresses. Private, loopback, and documentation ranges should not be sent to an external lookup service.

import ipaddress
import re
from collections import Counter
from pathlib import Path

LOG_LINE = re.compile(
    r'^(?P<ip>\S+) \S+ \S+ \[[^]]+\] "(?P<method>\S+) '
    r'(?P<path>\S+) [^"]+" (?P<status>\d{3}) (?P<size>\S+)'
)


def read_public_ip_counts(log_path: str) -> Counter[str]:
    counts: Counter[str] = Counter()

    for line in Path(log_path).read_text(errors="replace").splitlines():
        match = LOG_LINE.match(line)
        if not match:
            continue

        try:
            address = ipaddress.ip_address(match.group("ip"))
        except ValueError:
            continue

        if not address.is_global:
            continue

        counts[address.compressed] += 1

    return counts

If Nginx sits behind a reverse proxy, verify that the first field contains the trusted client address rather than the proxy address. Do not blindly log the first value in X-Forwarded-For; only accept forwarding headers from proxies you operate.

Look up each unique IP once

The synchronous Actor endpoint fits a batch job because each call returns its dataset item directly. The function below applies a timeout and distinguishes an empty result from an HTTP failure.

import os
import requests

APIFY_TOKEN = os.environ.get("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~ip-geolocation-lookup"
LOOKUP_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def geolocate(ip_address: str) -> dict:
    response = requests.post(
        LOOKUP_URL,
        params={"token": APIFY_TOKEN},
        json={"ip": ip_address},
        timeout=60,
    )
    response.raise_for_status()

    items = response.json()
    if not isinstance(items, list) or not items:
        return {"ip": ip_address, "lookup_status": "not_found"}

    item = items[0]
    return {
        "ip": ip_address,
        "country_code": item.get("countryCode"),
        "country": item.get("country"),
        "region": item.get("region"),
        "city": item.get("city"),
        "timezone": item.get("timezone"),
        "isp": item.get("isp"),
        "organization": item.get("org"),
        "asn": item.get("as"),
        "lookup_status": "ok",
    }

The token belongs in an environment variable or secret manager. Keeping YOUR_APIFY_TOKEN as the fallback makes accidental production execution fail clearly rather than hiding a real credential in source.

Add a durable lookup cache

A daily log may contain millions of requests but only thousands of unique addresses. The same addresses often return on following days, so a SQLite cache cuts both runtime and API calls.

IP ownership and routing change, which means cache entries need an expiry. Seven days is reasonable for a traffic report; fraud controls or account security may need fresher data.

import json
import sqlite3
from datetime import datetime, timedelta, timezone

CACHE_TTL = timedelta(days=7)


def open_cache(path: str = "ip-enrichment.sqlite3") -> sqlite3.Connection:
    connection = sqlite3.connect(path)
    connection.execute("""
        CREATE TABLE IF NOT EXISTS ip_cache (
            ip TEXT PRIMARY KEY,
            looked_up_at TEXT NOT NULL,
            payload TEXT NOT NULL
        )
    """)
    return connection


def cached_geolocate(connection: sqlite3.Connection, ip_address: str) -> dict:
    row = connection.execute(
        "SELECT looked_up_at, payload FROM ip_cache WHERE ip = ?",
        (ip_address,),
    ).fetchone()

    if row:
        looked_up_at = datetime.fromisoformat(row[0])
        if datetime.now(timezone.utc) - looked_up_at < CACHE_TTL:
            return json.loads(row[1])

    result = geolocate(ip_address)
    now = datetime.now(timezone.utc).isoformat()
    connection.execute(
        """
        INSERT INTO ip_cache (ip, looked_up_at, payload)
        VALUES (?, ?, ?)
        ON CONFLICT(ip) DO UPDATE SET
            looked_up_at = excluded.looked_up_at,
            payload = excluded.payload
        """,
        (ip_address, now, json.dumps(result, sort_keys=True)),
    )
    connection.commit()
    return result

I cache not_found results too. Otherwise a malformed or unsupported address gets retried in every run. A shorter TTL for failures is a useful refinement if temporary upstream gaps are common.

Build a network-level report

Exact city counts look impressive, but IP geolocation is not GPS. Country and network ownership are usually more appropriate for aggregate operations work. I keep the city field for investigation and summarize normal reports by country and ASN.

import csv
from collections import defaultdict


def build_traffic_report(log_path: str, output_path: str) -> None:
    ip_counts = read_public_ip_counts(log_path)
    totals = defaultdict(lambda: {"requests": 0, "unique_ips": 0})

    with open_cache() as connection:
        for ip_address, request_count in ip_counts.items():
            geo = cached_geolocate(connection, ip_address)
            country = geo.get("country_code") or "UNKNOWN"
            asn = geo.get("asn") or "UNKNOWN"
            key = (country, asn)
            totals[key]["requests"] += request_count
            totals[key]["unique_ips"] += 1

    rows = sorted(
        (
            {
                "country": country,
                "asn": asn,
                "requests": values["requests"],
                "unique_ips": values["unique_ips"],
            }
            for (country, asn), values in totals.items()
        ),
        key=lambda row: row["requests"],
        reverse=True,
    )

    with open(output_path, "w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle,
            fieldnames=("country", "asn", "requests", "unique_ips"),
        )
        writer.writeheader()
        writer.writerows(rows)

This output works with a spreadsheet, a scheduled email, or an ingestion job for Grafana. It also stays inspectable when somebody asks why a network was flagged.

Flag changes against your own baseline

A fixed list of suspicious countries ages badly and can block legitimate users. For operational review, compare each day's distribution with a rolling baseline and flag large changes rather than assigning guilt to a location.

def traffic_share(rows: list[dict], key: str, value: str) -> float:
    total = sum(int(row["requests"]) for row in rows)
    if total == 0:
        return 0.0
    matched = sum(
        int(row["requests"])
        for row in rows
        if row.get(key) == value
    )
    return matched / total


def is_share_spike(
    current_rows: list[dict],
    baseline_rows: list[dict],
    key: str,
    value: str,
    minimum_requests: int = 500,
) -> bool:
    current_requests = sum(
        int(row["requests"])
        for row in current_rows
        if row.get(key) == value
    )
    current = traffic_share(current_rows, key, value)
    baseline = traffic_share(baseline_rows, key, value)

    return current_requests >= minimum_requests and current >= max(0.05, baseline * 3)

A spike should open an investigation, not automatically add a firewall rule. Check paths, status codes, user agents, request rates, and authentication outcomes before taking action. Large cloud ASNs host both normal services and abusive automation.

Cost and data handling

The IP Geolocation Lookup is priced at $0.001 per run, so enriching 10,000 previously unseen IPs costs $10 in Actor charges. Apify platform usage may also apply. With a seven-day cache, repeat visitors do not create another lookup on every report run.

IP addresses can be personal data depending on jurisdiction and context. Set a retention period, restrict access to the raw cache, and aggregate before sending reports broadly. If the report only needs country and ASN totals, delete or rotate the raw address mapping after the operational window closes.

Run the job on rotated logs so each file is stable, write the CSV to a temporary filename, and rename it only after every lookup succeeds. That keeps dashboards from reading a half-written report and gives retries a clean input.

How to Build an Auditable Swedish Supplier Check with Python

Cizze R — Wed, 29 Jul 2026 12:00:12 +0000

How to Build an Auditable Swedish Supplier Check with Python

A supplier form that accepts a company name and moves on is easy to build, but it leaves someone doing registry checks by hand later. I wanted the onboarding service to verify the organization number, preserve what it found, and send ambiguous cases to a human instead of silently approving them.

The Swedish Company Registry handles the registry lookup, while Python handles the less glamorous parts that make the result useful: input cleanup, exact matching, review rules, and a small audit trail.

Treat the organization number as an identifier

Swedish organization numbers are commonly written as 556703-7482, but users also paste spaces, omit the hyphen, or include a Swedish prefix. Normalize the value before querying or comparing it, then retain the formatted version for display.

import re


def normalize_org_number(value: str) -> str:
    digits = re.sub(r"\D", "", value)

    # A Swedish VAT number can arrive as SE + 10 digits + 01.
    if digits.startswith("46") and len(digits) == 12:
        digits = digits[2:]
    if len(digits) == 12 and digits.endswith("01"):
        digits = digits[:-2]

    if len(digits) != 10:
        raise ValueError("Expected a 10-digit Swedish organization number")

    return digits


def display_org_number(value: str) -> str:
    digits = normalize_org_number(value)
    return f"{digits[:6]}-{digits[6:]}"

Normalization is not verification. It only gives every later step one representation to compare. The registry response is still the authority for whether that number exists and which company it identifies.

Run the registry lookup synchronously

For an onboarding request, waiting for a short Actor run is simpler than creating a run and polling it yourself. Apify's synchronous dataset endpoint returns the resulting items in the same HTTP response.

import requests

APIFY_TOKEN = "YOUR_APIFY_TOKEN"
ACTOR_ID = "weeknds~swedish-company-registry"
ACTOR_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


class RegistryLookupError(RuntimeError):
    pass


def search_registry(query: str, max_results: int = 10) -> list[dict]:
    response = requests.post(
        ACTOR_URL,
        params={"token": APIFY_TOKEN},
        json={"searchQuery": query, "maxResults": max_results},
        timeout=90,
    )

    if response.status_code >= 500:
        raise RegistryLookupError("Registry lookup is temporarily unavailable")

    response.raise_for_status()
    items = response.json()
    if not isinstance(items, list):
        raise RegistryLookupError("Unexpected registry response")
    return items

Keep the token in an environment variable in a real service. YOUR_APIFY_TOKEN is only there to make the request shape obvious.

A company-name search can return several similarly named businesses. An organization-number search should be much narrower, but I still require an exact normalized match rather than trusting the first result.

def exact_company_match(org_number: str, results: list[dict]) -> dict | None:
    expected = normalize_org_number(org_number)

    for company in results:
        candidate = company.get("org_number", "")
        try:
            if normalize_org_number(candidate) == expected:
                return company
        except ValueError:
            continue

    return None

Return a review decision, not a fake risk score

A registry check can confirm identity and expose useful status fields. It cannot tell you whether a supplier is trustworthy. I use three outcomes: verified, review, and not_found.

from datetime import date, datetime, timedelta


def parse_registry_date(value: str | None) -> date | None:
    if not value:
        return None
    try:
        return datetime.strptime(value, "%Y-%m-%d").date()
    except ValueError:
        return None


def assess_supplier(org_number: str) -> dict:
    formatted = display_org_number(org_number)
    results = search_registry(formatted, max_results=5)
    company = exact_company_match(formatted, results)

    if company is None:
        return {
            "decision": "not_found",
            "org_number": formatted,
            "reasons": ["No exact organization-number match"],
        }

    reasons = []
    status = company.get("status")
    registered = parse_registry_date(company.get("registration_date"))

    if status != "Registrerat":
        reasons.append(f"Registry status requires review: {status or 'missing'}")

    if registered and registered > date.today() - timedelta(days=90):
        reasons.append("Company was registered within the last 90 days")

    required = ("name", "org_number", "address")
    missing = [field for field in required if not company.get(field)]
    if missing:
        reasons.append("Missing registry fields: " + ", ".join(missing))

    return {
        "decision": "review" if reasons else "verified",
        "org_number": formatted,
        "company": {
            "name": company.get("name"),
            "legal_form": company.get("legal_form"),
            "status": status,
            "address": company.get("address"),
            "county": company.get("county"),
            "registration_date": company.get("registration_date"),
        },
        "reasons": reasons,
    }

The 90-day rule is an example review policy, not a claim that a new company is suspicious. Make thresholds explicit so compliance or procurement can change them without rewriting the integration.

Save the evidence you actually used

Registry data changes. If someone asks why a supplier passed six months ago, the current record alone does not answer the question. Store the decision, the selected fields, and the lookup timestamp.

SQLite works well for a small internal service and keeps the example deployable without another database.

import json
import sqlite3
from datetime import datetime, timezone


def open_audit_db(path: str = "supplier-checks.sqlite3") -> sqlite3.Connection:
    connection = sqlite3.connect(path)
    connection.execute("""
        CREATE TABLE IF NOT EXISTS supplier_checks (
            id INTEGER PRIMARY KEY,
            org_number TEXT NOT NULL,
            checked_at TEXT NOT NULL,
            decision TEXT NOT NULL,
            result_json TEXT NOT NULL
        )
    """)
    return connection


def check_and_record(org_number: str, db_path: str = "supplier-checks.sqlite3") -> dict:
    result = assess_supplier(org_number)
    checked_at = datetime.now(timezone.utc).isoformat()

    with open_audit_db(db_path) as connection:
        connection.execute(
            """
            INSERT INTO supplier_checks
                (org_number, checked_at, decision, result_json)
            VALUES (?, ?, ?, ?)
            """,
            (
                result["org_number"],
                checked_at,
                result["decision"],
                json.dumps(result, ensure_ascii=False, sort_keys=True),
            ),
        )

    return {**result, "checked_at": checked_at}

Avoid storing more personal data than the workflow needs. Company records can include officers, but a basic supplier identity check may only need the registered name, organization number, status, and address. Data minimization makes the audit record easier to defend and maintain.

Put the check behind a narrow API

Your purchasing app should not know Actor input fields or registry response details. A small service boundary lets the UI submit an organization number and receive a stable decision format.

from flask import Flask, jsonify, request

app = Flask(__name__)


@app.post("/supplier-checks")
def create_supplier_check():
    payload = request.get_json(silent=True) or {}

    try:
        result = check_and_record(payload.get("org_number", ""))
    except ValueError as exc:
        return jsonify({"error": str(exc)}), 400
    except (requests.RequestException, RegistryLookupError):
        return jsonify({"error": "Registry check unavailable"}), 503

    status_code = 200 if result["decision"] == "verified" else 202
    return jsonify(result), status_code

A 202 for review cases prevents the caller from confusing a valid API response with automatic approval. Network failures return 503, so they can be retried instead of becoming false rejections.

Cost, caching, and rechecks

The Swedish Company Registry is priced at $0.003 per search, so 1,000 onboarding checks cost $3 in Actor charges. Apify platform usage can also apply, and the Actor page is the right place to confirm current pricing before a bulk run.

Do not cache forever just to save a fraction of a cent. Cache repeated requests during one onboarding session, then run a fresh check before approval if the saved evidence is older than your policy allows. A 30-day recheck window may work for routine procurement, while a payment change or compliance event should trigger an immediate lookup.

The audit row should remain immutable even after a recheck. Insert a new row, compare the new status with the previous one, and route any change in registered name, address, or status back to review.

I Replaced Browser-Based Social Card Checks with One Metadata API

Cizze R — Mon, 27 Jul 2026 12:00:28 +0000

I Replaced Browser-Based Social Card Checks with One Metadata API

I used to open a browser in CI just to confirm that release pages had titles, descriptions, and social images. The checks were slow, the browser image needed regular maintenance, and a passing page render still did not tell me whether the Open Graph tags were complete.

I replaced that step with the Link Preview & Metadata Extractor and a small Python policy checker. It fetches the HTML over HTTP, returns the metadata and response details as structured JSON, and leaves the project-specific pass or fail rules in my code.

The browser was testing the wrong layer

A browser is useful when the question is whether a page renders correctly. My actual question was narrower: what will a link unfurler or crawler find in the document metadata?

For that check I care about:

A successful HTML response
A usable title and description
An absolute social image URL
A canonical URL on the expected host
An Open Graph type
Twitter Card metadata where the product supports X sharing
Robots directives that do not accidentally block indexing

None of those requires screenshots or layout inspection. The actor uses HTTP and parses the document metadata directly, including Open Graph fields, Twitter Cards, favicons, language, canonical URLs, all meta tags, status code, content type, and response timing.

That also creates a better failure boundary. DNS, SSL, timeout, HTTP, and non-HTML errors come back as structured errors instead of a generic browser navigation failure.

Fetch the exact deployed URL

The audit runs after deployment against the public URL, not against rendered template source. That catches reverse proxy redirects, missing production environment variables, and CDN responses that local tests never see.

import os
from typing import Any

import requests

APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~link-preview-metadata-extractor"
RUN_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def extract_metadata(url: str) -> dict[str, Any]:
    response = requests.post(
        RUN_URL,
        params={"token": APIFY_TOKEN},
        json={
            "url": url,
            "includeFavicon": True,
            "userAgent": "Release-Metadata-Audit/1.0",
            "timeout": 20,
        },
        timeout=75,
    )
    response.raise_for_status()
    items = response.json()
    if len(items) != 1:
        raise RuntimeError(f"Expected one result for {url}, got {len(items)}")
    return items[0]

A custom user agent makes the audit identifiable in server logs. It should describe the client honestly, not impersonate a search crawler.

Flatten the response before applying policy

Metadata extractors can group Open Graph and HTTP data into nested objects. A recursive flattener lets the policy work with either nested values or literal keys such as og:title.

from collections.abc import Mapping
from typing import Any


def flatten(value: Mapping[str, Any], prefix: str = "") -> dict[str, Any]:
    output = {}
    for key, child in value.items():
        path = f"{prefix}.{key}" if prefix else key
        output[path] = child
        if isinstance(child, Mapping):
            output.update(flatten(child, path))
    return output


def find(item: dict[str, Any], *keys: str) -> Any:
    fields = flatten(item)
    for key in keys:
        for path, value in fields.items():
            if path == key or path.endswith(f".{key}"):
                if value not in (None, "", []):
                    return value
    return None

I keep this adapter separate from the rules. If the actor adds a new output field, the audit can use it without changing the network layer.

Turn metadata expectations into explicit rules

A release check should produce actionable messages. "Metadata invalid" sends someone back to the logs, while "og:image must be an absolute HTTPS URL" points at the fix.

from dataclasses import dataclass
from urllib.parse import urlparse


@dataclass(frozen=True)
class Finding:
    severity: str
    field: str
    message: str


def audit_page(item: dict, expected_host: str) -> list[Finding]:
    findings = []

    if item.get("error"):
        return [Finding(
            "error",
            item.get("error_type", "fetch"),
            item["error"],
        )]

    status = find(item, "status_code")
    content_type = str(find(item, "content_type") or "")
    title = find(item, "og:title", "title")
    description = find(item, "og:description", "description")
    image = find(item, "og:image")
    canonical = find(item, "canonical_url", "canonical", "og:url")
    og_type = find(item, "og:type")
    robots = str(find(item, "robots") or "").lower()

    if status and not 200 <= int(status) < 300:
        findings.append(Finding("error", "status_code", f"HTTP {status}"))
    if content_type and "text/html" not in content_type:
        findings.append(Finding("error", "content_type", content_type))
    if not title or not 10 <= len(str(title)) <= 70:
        findings.append(Finding("error", "title", "Use a 10-70 character title"))
    if not description or not 50 <= len(str(description)) <= 200:
        findings.append(Finding(
            "warning",
            "description",
            "Use a 50-200 character description",
        ))
    if not image or urlparse(str(image)).scheme != "https":
        findings.append(Finding(
            "error",
            "og:image",
            "Use an absolute HTTPS image URL",
        ))
    if not canonical:
        findings.append(Finding("warning", "canonical", "Canonical URL is missing"))
    elif urlparse(str(canonical)).hostname != expected_host:
        findings.append(Finding(
            "error",
            "canonical",
            f"Expected canonical host {expected_host}",
        ))
    if not og_type:
        findings.append(Finding("warning", "og:type", "Open Graph type is missing"))
    if "noindex" in robots:
        findings.append(Finding("error", "robots", "Page is marked noindex"))

    return findings

The length ranges are policy, not universal laws. I use them to catch obviously empty or accidentally huge values, not to pretend every social platform truncates at the same character.

Audit a release manifest instead of crawling everything

A full-site crawl on every release wastes time and makes ownership unclear. My deployment exports a short manifest of URLs affected by the release, then the audit checks only those pages.

import json
from pathlib import Path


def load_release_urls(path: Path) -> list[dict[str, str]]:
    payload = json.loads(path.read_text(encoding="utf-8"))
    urls = payload.get("urls", [])
    if not urls:
        raise ValueError("Release manifest contains no URLs")
    return urls


def audit_release(manifest_path: Path) -> int:
    error_count = 0

    for entry in load_release_urls(manifest_path):
        url = entry["url"]
        expected_host = entry["expected_host"]

        try:
            item = extract_metadata(url)
            findings = audit_page(item, expected_host)
        except (requests.RequestException, RuntimeError, ValueError) as error:
            findings = [Finding("error", "request", str(error))]

        if not findings:
            print(f"PASS {url}")
            continue

        for finding in findings:
            print(
                f"{finding.severity.upper()} {url} "
                f"[{finding.field}] {finding.message}"
            )
            if finding.severity == "error":
                error_count += 1

    return error_count


if __name__ == "__main__":
    failures = audit_release(Path("release-urls.json"))
    raise SystemExit(1 if failures else 0)

The manifest stays simple and reviewable:

{
  "urls": [
    {
      "url": "https://docs.example.com/python/cache-invalidation",
      "expected_host": "docs.example.com"
    },
    {
      "url": "https://example.com/releases/summer-update",
      "expected_host": "example.com"
    }
  ]
}

Warnings appear in the job output but do not block the release. Errors return a non-zero exit code. That distinction stopped minor description preferences from becoming production incidents while still blocking broken canonicals, failed fetches, insecure image URLs, and accidental noindex directives.

Save evidence for failed checks

The worst version of this job only printed a summary. By the time someone investigated, a redeploy had changed the page and the original response was gone.

Now the audit writes the actor item whenever a page fails. The artifact contains the metadata and HTTP details observed at release time, without requiring another request.

import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path

ARTIFACT_DIR = Path("metadata-audit-artifacts")


def save_failure_artifact(url: str, item: dict, findings: list[Finding]) -> Path:
    ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)
    digest = hashlib.sha256(url.encode()).hexdigest()[:12]
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    path = ARTIFACT_DIR / f"{timestamp}-{digest}.json"
    path.write_text(
        json.dumps(
            {
                "url": url,
                "findings": [finding.__dict__ for finding in findings],
                "metadata": item,
            },
            indent=2,
            sort_keys=True,
        ),
        encoding="utf-8",
    )
    return path

Treat these files as operational artifacts, not permanent content. A 30-day retention window is usually enough to diagnose a release without building an unbounded archive.

What the switch changed

The audit no longer downloads a browser image or waits for client rendering. It tests the metadata layer directly, and each rule names the field that failed. Browser-based visual checks still exist for pages where layout matters, but they are no longer the default tool for validating social cards.

The Link Preview & Metadata Extractor is currently listed as pay per usage on Apify. The live price depends on platform resource consumption and the account plan, so the job should check the current store pricing before turning a small release audit into a large crawl.

Run the audit only after the deployment URL is reachable, keep the token in the CI secret store, and cap retries for network errors. A timeout can deserve one retry; a noindex directive, non-HTML response, or wrong canonical host needs a code or configuration change, not another request.

How to Build a URL Metadata Cache with Python and SQLite

Cizze R — Fri, 24 Jul 2026 12:00:22 +0000

How to Build a URL Metadata Cache with Python and SQLite

Fetching metadata when a user opens a page makes link previews feel slow and turns every refresh into another network request. A small SQLite cache moves that work to ingestion time, gives you predictable reads, and keeps the raw response available when your UI needs a new field later.

The Link Preview & Metadata Extractor handles the remote fetch and parses Open Graph, Twitter Cards, canonical URLs, favicons, language, and HTTP details. Python only needs to call the actor, normalize the useful fields, and decide when an entry is stale.

Start with the synchronous actor endpoint

The actor takes one required url. Optional inputs let you include favicon discovery, set a custom user agent, and choose a request timeout between 1 and 120 seconds.

For an ingestion worker, Apify's synchronous dataset endpoint is simpler than starting a run and polling it. The response is the dataset item, so one function can return either metadata or a structured actor error.

import os
from typing import Any

import requests

APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~link-preview-metadata-extractor"
RUN_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def fetch_metadata(url: str, timeout_seconds: int = 15) -> dict[str, Any]:
    response = requests.post(
        RUN_URL,
        params={"token": APIFY_TOKEN},
        json={
            "url": url,
            "includeFavicon": True,
            "timeout": timeout_seconds,
        },
        timeout=timeout_seconds + 45,
    )
    response.raise_for_status()
    items = response.json()
    if not items:
        raise RuntimeError(f"No metadata item returned for {url}")

    item = items[0]
    if item.get("error"):
        error_type = item.get("error_type", "actor_error")
        raise RuntimeError(f"{error_type}: {item['error']}")
    return item

Keep APIFY_TOKEN in the environment rather than in the database or source file. The HTTP timeout is intentionally longer than the actor's page timeout because the platform also needs time to start the run and return its dataset.

Preserve raw data and index the fields you query

Metadata formats change more often than database schemas should. One site may provide og:title, another only a regular title, and a third may omit both but include Twitter Card data.

The practical compromise is to store the full actor item as JSON while copying a few frequently queried values into columns. The helper below flattens nested dictionaries, so it works whether a field is returned at the top level or inside an Open Graph object.

import json
from collections.abc import Mapping
from typing import Any


def flatten_mapping(value: Mapping[str, Any], prefix: str = "") -> dict[str, Any]:
    flattened = {}
    for key, child in value.items():
        path = f"{prefix}.{key}" if prefix else key
        flattened[path] = child
        if isinstance(child, Mapping):
            flattened.update(flatten_mapping(child, path))
    return flattened


def first_value(item: dict[str, Any], *names: str) -> Any:
    flat = flatten_mapping(item)
    for name in names:
        for path, value in flat.items():
            if path == name or path.endswith(f".{name}"):
                if value not in (None, "", []):
                    return value
    return None


def normalize_item(requested_url: str, item: dict[str, Any]) -> dict[str, Any]:
    canonical = first_value(item, "canonical_url", "canonical", "og:url")
    return {
        "requested_url": requested_url,
        "canonical_url": canonical or requested_url,
        "title": first_value(item, "og:title", "twitter:title", "title"),
        "description": first_value(
            item,
            "og:description",
            "twitter:description",
            "description",
        ),
        "image_url": first_value(item, "og:image", "twitter:image"),
        "favicon_url": first_value(item, "favicon", "favicon_url"),
        "status_code": first_value(item, "status_code"),
        "raw_json": json.dumps(item, sort_keys=True),
    }

Saving the raw object matters. If you later decide to display language, author, robots directives, or response timing, you can backfill columns from cached responses instead of re-fetching every URL.

Create a cache that survives restarts

SQLite is enough for a personal knowledge tool, internal dashboard, or small content product. Use the requested URL as the lookup key and keep the canonical URL separately for deduplication.

import sqlite3
from pathlib import Path

DATABASE_PATH = Path("link-metadata.sqlite3")


def connect() -> sqlite3.Connection:
    database = sqlite3.connect(DATABASE_PATH)
    database.row_factory = sqlite3.Row
    database.execute("PRAGMA journal_mode=WAL")
    database.execute("PRAGMA busy_timeout=5000")
    database.execute(
        """
        CREATE TABLE IF NOT EXISTS link_metadata (
            requested_url TEXT PRIMARY KEY,
            canonical_url TEXT NOT NULL,
            title TEXT,
            description TEXT,
            image_url TEXT,
            favicon_url TEXT,
            status_code INTEGER,
            raw_json TEXT NOT NULL,
            fetched_at TEXT NOT NULL
        )
        """
    )
    database.execute(
        "CREATE INDEX IF NOT EXISTS metadata_canonical_idx "
        "ON link_metadata(canonical_url)"
    )
    return database

WAL mode lets readers continue while an ingestion worker writes. The busy timeout prevents a short overlapping transaction from immediately failing with a locked database error.

Upsert atomically after a successful extraction

Do not replace a good cache entry before the actor returns valid data. If the target times out or returns non-HTML content, the previous preview can remain available while the failure goes to your logs.

from datetime import datetime, timezone


def save_metadata(database: sqlite3.Connection, record: dict) -> None:
    database.execute(
        """
        INSERT INTO link_metadata (
            requested_url, canonical_url, title, description,
            image_url, favicon_url, status_code, raw_json, fetched_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        ON CONFLICT(requested_url) DO UPDATE SET
            canonical_url = excluded.canonical_url,
            title = excluded.title,
            description = excluded.description,
            image_url = excluded.image_url,
            favicon_url = excluded.favicon_url,
            status_code = excluded.status_code,
            raw_json = excluded.raw_json,
            fetched_at = excluded.fetched_at
        """,
        (
            record["requested_url"],
            record["canonical_url"],
            record["title"],
            record["description"],
            record["image_url"],
            record["favicon_url"],
            record["status_code"],
            record["raw_json"],
            datetime.now(timezone.utc).isoformat(),
        ),
    )
    database.commit()


def enrich_url(database: sqlite3.Connection, url: str) -> dict:
    item = fetch_metadata(url)
    record = normalize_item(url, item)
    save_metadata(database, record)
    return record

This transaction is small, but it gives the cache a useful guarantee: readers see either the previous complete record or the new complete record, never half of each.

Refresh only stale entries

Metadata does not need to be fetched on every request. News links may deserve a daily refresh, while documentation and reference pages can sit for a week or longer.

from datetime import datetime, timedelta, timezone


def get_cached(
    database: sqlite3.Connection,
    url: str,
    max_age: timedelta = timedelta(days=7),
) -> dict | None:
    row = database.execute(
        "SELECT * FROM link_metadata WHERE requested_url = ?",
        (url,),
    ).fetchone()
    if row is None:
        return None

    fetched_at = datetime.fromisoformat(row["fetched_at"])
    if datetime.now(timezone.utc) - fetched_at > max_age:
        return None
    return dict(row)


def metadata_for_url(database: sqlite3.Connection, url: str) -> dict:
    cached = get_cached(database, url)
    return cached if cached is not None else enrich_url(database, url)

In a web app, keep stale refreshes off the request path. Return the old record immediately and put the URL on a worker queue. The next request gets fresh metadata without making the current user wait for an external fetch.

Deduplicate tracking URLs by canonical URL

Requested URLs still need separate cache entries because they may redirect differently. For your application model, however, two records with the same canonical URL usually represent the same resource.

def canonical_duplicates(database: sqlite3.Connection) -> list[dict]:
    rows = database.execute(
        """
        SELECT canonical_url,
               COUNT(*) AS url_count,
               GROUP_CONCAT(requested_url, '\n') AS requested_urls
        FROM link_metadata
        GROUP BY canonical_url
        HAVING COUNT(*) > 1
        ORDER BY url_count DESC
        """
    ).fetchall()
    return [dict(row) for row in rows]

This catches campaign parameters, URL shorteners, and alternate share links after the destination declares a canonical URL. Do not blindly merge when the canonical is missing. Falling back to the requested URL is safer than guessing that two similar paths are equivalent.

Cost and operating limits

The Link Preview & Metadata Extractor is currently listed on Apify as pay per usage. Its store pricing depends on the platform resources consumed and the account plan, so check the live pricing page before estimating a large import. Because it fetches HTML over HTTP without a headless browser, it avoids the browser startup work that makes screenshot-based extraction heavier.

Keep a modest worker pool for batch imports and retry only transient network failures. A permanent non_html or HTTP error should be recorded for review rather than retried in a tight loop. Back up the SQLite file with its WAL checkpointed, and retain raw_json even if the preview UI currently uses only three fields.

I Stopped Running Playwright in Cron and Built a Screenshot Archive Instead

Cizze R — Wed, 22 Jul 2026 12:00:38 +0000

I Stopped Running Playwright in Cron and Built a Screenshot Archive Instead

Taking one website screenshot with Playwright is easy. Keeping a scheduled screenshot job healthy for months is where the work starts: browser updates, missing system libraries, hanging pages, and a growing pile of images with names nobody can search.

I moved the browser part to the Website Screenshot API and kept the useful part locally: scheduling, file naming, manifests, retention, and failure reporting. The result is a small Python archive that captures the same pages in desktop, mobile, and dark mode without running Chromium on the cron host.

The archive, not the screenshot, is the feature

My first version saved files like screenshot.png, screenshot-1.png, and final-new.png. That was fine until I needed to answer a basic question: what did the pricing page look like on mobile last Tuesday?

A useful archive needs more structure. Each capture should record:

The requested URL
The viewport preset and actual dimensions
Whether dark mode and full-page capture were enabled
The actor's timestamp and success status
The remote image URL
The local filename and SHA-256 checksum

The checksum catches duplicate images even if they were created by separate runs. A JSON Lines manifest makes the archive searchable without introducing a database.

Use the synchronous dataset endpoint

The actor accepts a URL, viewport preset, optional custom dimensions, full-page mode, a delay, PNG or JPEG output, and dark-mode emulation. It stores the binary in an Apify key-value store and returns a public screenshotUrl in the dataset item.

import os
from typing import Any

import requests

APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~website-screenshot-api"
RUN_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def request_screenshot(
    url: str,
    viewport: str = "desktop",
    *,
    full_page: bool = True,
    dark_mode: bool = False,
    delay_ms: int = 1500,
    image_format: str = "png",
) -> dict[str, Any]:
    payload = {
        "url": url,
        "viewport": viewport,
        "fullPage": full_page,
        "delay": delay_ms,
        "format": image_format,
        "darkMode": dark_mode,
    }
    response = requests.post(
        RUN_URL,
        params={"token": APIFY_TOKEN},
        json=payload,
        timeout=120,
    )
    response.raise_for_status()
    items = response.json()
    if not items:
        raise RuntimeError(f"No screenshot result returned for {url}")

    result = items[0]
    if not result.get("success"):
        raise RuntimeError(result.get("error") or "Screenshot failed")
    return result

A short delay is worth keeping for pages with client-rendered content or lazy-loaded images. It is not a guarantee that every network request has settled, but it removes many half-rendered captures without adding browser logic to the local script.

Give every image a deterministic path

The archive path should communicate what was captured before you open the file. I use the hostname, date, viewport, color mode, and a short URL hash. The hash prevents /pricing and /docs/pricing from colliding after slug cleanup.

import hashlib
import re
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse

ARCHIVE_ROOT = Path("screenshot-archive")


def slugify(value: str) -> str:
    cleaned = re.sub(r"[^a-z0-9]+", "-", value.lower())
    return cleaned.strip("-") or "page"


def archive_path(
    url: str,
    viewport: str,
    dark_mode: bool,
    image_format: str,
    captured_at: datetime,
) -> Path:
    parsed = urlparse(url)
    host = slugify(parsed.netloc)
    page = slugify(parsed.path or "home")
    url_hash = hashlib.sha256(url.encode()).hexdigest()[:8]
    mode = "dark" if dark_mode else "light"
    day = captured_at.strftime("%Y/%m/%d")
    filename = f"{page}-{url_hash}-{viewport}-{mode}.{image_format}"
    return ARCHIVE_ROOT / host / day / filename

The path is deterministic within a day. If the job runs twice, the newer capture replaces the earlier one rather than quietly creating duplicate filenames. Add hours and minutes if intraday history matters.

Download safely and calculate a checksum

Do not write the response directly to its final path. A network interruption can otherwise leave a truncated file that looks valid to later scripts.

import hashlib
from pathlib import Path


def download_image(image_url: str, destination: Path) -> str:
    destination.parent.mkdir(parents=True, exist_ok=True)
    temporary = destination.with_suffix(destination.suffix + ".part")
    digest = hashlib.sha256()

    with requests.get(image_url, stream=True, timeout=60) as response:
        response.raise_for_status()
        with temporary.open("wb") as output:
            for chunk in response.iter_content(chunk_size=1024 * 128):
                if not chunk:
                    continue
                output.write(chunk)
                digest.update(chunk)

    temporary.replace(destination)
    return digest.hexdigest()

A .part file also makes interrupted jobs easy to spot and clean up. If the download completes, replace() is atomic on the same filesystem.

Append a searchable manifest

JSON Lines works well here because each record is independent. You can append one capture without reading and rewriting the full history.

import json
from pathlib import Path

MANIFEST_PATH = ARCHIVE_ROOT / "manifest.jsonl"


def append_manifest(record: dict) -> None:
    MANIFEST_PATH.parent.mkdir(parents=True, exist_ok=True)
    with MANIFEST_PATH.open("a", encoding="utf-8") as manifest:
        manifest.write(json.dumps(record, sort_keys=True) + "\n")

The actor returns its own timestamp, viewport dimensions, format, full-page setting, dark-mode setting, and image URL. Preserve those values instead of reconstructing them from the request, because the response reflects what actually ran.

from datetime import datetime, timezone


def capture_to_archive(
    url: str,
    viewport: str,
    dark_mode: bool,
) -> dict:
    result = request_screenshot(
        url,
        viewport,
        full_page=True,
        dark_mode=dark_mode,
        delay_ms=2000,
        image_format="png",
    )

    captured_at = datetime.now(timezone.utc)
    destination = archive_path(
        url=url,
        viewport=viewport,
        dark_mode=dark_mode,
        image_format=result["format"],
        captured_at=captured_at,
    )
    checksum = download_image(result["screenshotUrl"], destination)

    record = {
        "url": result["url"],
        "captured_at": result.get("timestamp", captured_at.isoformat()),
        "viewport": result["viewport"],
        "format": result["format"],
        "full_page": result["fullPage"],
        "dark_mode": result["darkMode"],
        "remote_url": result["screenshotUrl"],
        "local_path": str(destination),
        "sha256": checksum,
    }
    append_manifest(record)
    return record

Capture a useful device matrix

A full matrix of every viewport and color mode becomes expensive and difficult to review. I settled on three captures per page: desktop light, mobile light, and desktop dark. That catches most responsive and theme problems without producing six nearly identical images.

from concurrent.futures import ThreadPoolExecutor, as_completed

CAPTURE_PROFILES = [
    ("desktop", False),
    ("mobile", False),
    ("desktop", True),
]


def archive_pages(urls: list[str], workers: int = 3) -> list[dict]:
    jobs = [
        (url, viewport, dark_mode)
        for url in urls
        for viewport, dark_mode in CAPTURE_PROFILES
    ]
    records = []

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {
            pool.submit(capture_to_archive, *job): job
            for job in jobs
        }
        for future in as_completed(futures):
            url, viewport, dark_mode = futures[future]
            try:
                record = future.result()
                records.append(record)
                print(f"saved {record['local_path']}")
            except Exception as error:
                mode = "dark" if dark_mode else "light"
                print(f"failed {url} {viewport} {mode}: {error}")

    return records

Keep the worker count modest. The goal is to shorten a portfolio run, not launch a burst that makes target sites or your Apify account unhappy.

Skip unchanged screenshots

Once checksums are in the manifest, duplicate detection is cheap. Load the most recent checksum for each URL and profile, then delete a new local image if it is byte-for-byte identical.

import json


def latest_checksums() -> dict[tuple[str, str, bool], str]:
    latest = {}
    if not MANIFEST_PATH.exists():
        return latest

    for line in MANIFEST_PATH.read_text(encoding="utf-8").splitlines():
        record = json.loads(line)
        viewport = record["viewport"]
        width = viewport["width"]
        profile = "mobile" if width == 375 else "desktop"
        key = (record["url"], profile, record["dark_mode"])
        latest[key] = record["sha256"]
    return latest

Byte equality is intentionally strict. Dynamic timestamps, cookie banners, and rotating ads will produce different hashes even when the layout is fine. If you need tolerance for small visual changes, send the previous and current images to Screenshot Comparison Tool rather than inventing an image comparison threshold around SHA-256.

Cost and retention

The Website Screenshot API is currently listed as pay per usage on Apify. There is no actor-specific fixed subscription shown on its store page, so the final cost depends on the compute consumed by each Playwright run and your Apify plan. Full-page captures and longer delays generally do more work than viewport-only captures.

Storage usually becomes the bigger local concern. A simple policy is to keep daily images for 30 days, one image per week for six months, and one per month after that. Never delete the manifest rows when pruning files. Mark them with a local_deleted_at field in a compacted manifest so the historical capture record remains honest.

For cron, run the script once per day and redirect standard error to your alerting system. A failed capture should leave the previous archive untouched, while successful captures arrive through atomic .part file replacement.

How to Monitor SSL Certificates and DNS Changes with Python

Cizze R — Mon, 20 Jul 2026 12:00:54 +0000

How to Monitor SSL Certificates and DNS Changes with Python

SSL expiry alerts are useful, but they only cover one part of a domain's operational state. A changed nameserver, a new mail provider, or an unexpected subdomain can be just as important.

I wanted one scheduled job that could watch all of those signals without maintaining separate WHOIS, DNS, TLS, and Certificate Transparency clients. The Domain Intelligence Suite returns the four data sets in one run, which makes the monitoring code mostly a matter of storing snapshots and comparing them.

What the monitor should detect

A useful domain monitor needs to report changes, not dump a fresh wall of JSON every morning. For each domain, we will track:

SSL certificate expiry and issuer
A, AAAA, MX, NS, and TXT records
Registrar and registration expiry
Subdomains found in Certificate Transparency logs
Errors from individual lookup modules

That last item matters. WHOIS can fail while DNS and SSL still work. The actor keeps module failures isolated, so the monitor can use the good results instead of treating the entire check as failed.

The actor accepts a domain, an optional list of modules, an SSL port, DNS record types, and a subdomain limit. Protocols and paths are stripped automatically, but passing a clean hostname keeps your snapshot keys predictable.

Calling the actor synchronously

For a scheduled job, the synchronous dataset endpoint is convenient because the HTTP response contains the result item. There is no run polling loop to maintain.

import os
from typing import Any

import requests

APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~domain-intelligence-suite"
RUN_URL = (
    f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
    "run-sync-get-dataset-items"
)


def inspect_domain(domain: str) -> dict[str, Any]:
    payload = {
        "domain": domain,
        "modules": ["whois", "dns", "ssl", "subdomains"],
        "sslPort": 443,
        "dnsRecordTypes": ["A", "AAAA", "MX", "NS", "TXT"],
        "maxSubdomains": 500,
        "includeWildcardSubdomains": False,
    }
    response = requests.post(
        RUN_URL,
        params={"token": APIFY_TOKEN},
        json=payload,
        timeout=120,
    )
    response.raise_for_status()
    items = response.json()
    if not items:
        raise RuntimeError(f"No domain intelligence returned for {domain}")
    return items[0]

Keep the token in an environment variable in production. The fallback value is there so the example is runnable after replacing YOUR_APIFY_TOKEN, not as a recommendation to commit tokens.

Reduce the response to stable signals

Raw responses contain query timestamps and other values that change on every run. Comparing the entire item would create noise. Instead, normalize only the fields that should trigger an alert.

DNS MX records are objects, while A and NS records are usually strings. Sorting through a JSON representation gives us a stable comparison for either shape.

import json
from typing import Any


def stable_list(values: list[Any] | None) -> list[Any]:
    return sorted(
        values or [],
        key=lambda value: json.dumps(value, sort_keys=True),
    )


def module_error(module: dict[str, Any] | None) -> str | None:
    if not module:
        return "module missing from response"
    return module.get("error")


def normalize_snapshot(item: dict[str, Any]) -> dict[str, Any]:
    whois = item.get("whois", {})
    dns = item.get("dns", {})
    ssl = item.get("ssl", {})
    subdomains = item.get("subdomains", {})

    whois_data = whois.get("data") or {}
    records = dns.get("records") or {}
    certificate = ssl.get("certificate") or {}

    return {
        "domain": item["domain"],
        "whois": {
            "registrar": whois_data.get("registrar"),
            "expiration_date": whois_data.get("expiration_date"),
            "name_servers": stable_list(whois_data.get("name_servers")),
            "error": module_error(whois),
        },
        "dns": {
            record_type: stable_list(records.get(record_type))
            for record_type in ["A", "AAAA", "MX", "NS", "TXT"]
        } | {"error": module_error(dns)},
        "ssl": {
            "issuer": certificate.get("issuer"),
            "valid_to": certificate.get("valid_to"),
            "days_remaining": certificate.get("days_remaining"),
            "expired": certificate.get("expired"),
            "subject_alt_names": stable_list(
                certificate.get("subject_alt_names")
            ),
            "error": module_error(ssl),
        },
        "subdomains": {
            "names": stable_list(subdomains.get("subdomains")),
            "error": module_error(subdomains),
        },
    }

The actor discovers subdomains through crt.sh, so this is passive discovery from public Certificate Transparency logs. It does not prove that every hostname is live. Treat a new name as a review signal, then resolve or probe it separately if needed.

Save snapshots without a database

A JSON file per domain is enough for a small portfolio. Write through a temporary file so a stopped process cannot leave a half-written snapshot behind.

import json
from pathlib import Path

SNAPSHOT_DIR = Path("domain-snapshots")


def snapshot_path(domain: str) -> Path:
    safe_name = domain.lower().replace("/", "_")
    return SNAPSHOT_DIR / f"{safe_name}.json"


def load_previous(domain: str) -> dict | None:
    path = snapshot_path(domain)
    if not path.exists():
        return None
    return json.loads(path.read_text(encoding="utf-8"))


def save_snapshot(domain: str, snapshot: dict) -> None:
    SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True)
    destination = snapshot_path(domain)
    temporary = destination.with_suffix(".tmp")
    temporary.write_text(
        json.dumps(snapshot, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    temporary.replace(destination)

On the first run, the file becomes the baseline. Later runs compare against it.

Report only meaningful differences

A recursive diff keeps the comparison reusable. Lists are already normalized, so added DNS records and subdomains show up as clean old/new values.

from typing import Any


def diff_values(old: Any, new: Any, path: str = "") -> list[dict]:
    if isinstance(old, dict) and isinstance(new, dict):
        changes = []
        for key in sorted(set(old) | set(new)):
            child_path = f"{path}.{key}" if path else key
            if key not in old:
                changes.append(
                    {"field": child_path, "old": None, "new": new[key]}
                )
            elif key not in new:
                changes.append(
                    {"field": child_path, "old": old[key], "new": None}
                )
            else:
                changes.extend(diff_values(old[key], new[key], child_path))
        return changes

    if old != new:
        return [{"field": path, "old": old, "new": new}]
    return []


def urgent_ssl_warning(snapshot: dict, threshold_days: int = 30) -> str | None:
    ssl = snapshot["ssl"]
    if ssl["error"]:
        return f"SSL lookup failed: {ssl['error']}"
    days = ssl["days_remaining"]
    if days is not None and days <= threshold_days:
        return f"SSL certificate expires in {days} days"
    return None

One caveat: days_remaining changes daily, so it will appear in the diff every day. If you only want threshold alerts, remove that field from the general comparison and keep urgent_ssl_warning().

Run it across a domain portfolio

The final script reads one domain per line, continues when an individual lookup fails, and only replaces a baseline after a successful result.

import sys


def monitor_domains(domains: list[str]) -> int:
    alert_count = 0

    for domain in domains:
        try:
            current = normalize_snapshot(inspect_domain(domain))
            previous = load_previous(domain)

            if previous is None:
                print(f"{domain}: baseline created")
            else:
                changes = diff_values(previous, current)
                for change in changes:
                    print(
                        f"{domain}: {change['field']} changed "
                        f"from {change['old']!r} to {change['new']!r}"
                    )
                alert_count += len(changes)

            warning = urgent_ssl_warning(current)
            if warning:
                print(f"{domain}: {warning}", file=sys.stderr)
                alert_count += 1

            save_snapshot(domain, current)
        except (requests.RequestException, RuntimeError, ValueError) as error:
            print(f"{domain}: check failed: {error}", file=sys.stderr)
            alert_count += 1

    return alert_count


if __name__ == "__main__":
    domains = [
        line.strip()
        for line in Path("domains.txt").read_text().splitlines()
        if line.strip() and not line.startswith("#")
    ]
    raise SystemExit(1 if monitor_domains(domains) else 0)

That exit code is useful in cron, systemd timers, and CI jobs: zero means no changes, one means the output deserves attention. If daily days_remaining changes are enabled, switch the exit decision to count only selected fields.

Cost and scheduling

The Domain Intelligence Suite currently uses Apify's pay-per-usage model rather than a fixed subscription for the actor. The store describes a typical run as taking roughly 3 to 10 seconds, with the final charge depending on compute usage and your Apify plan. Running only the modules you need reduces both work and noise.

WHOIS and DNS rarely need minute-by-minute checks. A daily schedule is sensible for SSL and infrastructure monitoring, while a weekly schedule is usually enough for registration due diligence. Keep maxSubdomains conservative for routine runs, and raise it only when you are investigating a larger domain.

How to Generate Custom QR Codes with Python

Cizze R — Fri, 17 Jul 2026 12:00:28 +0000

How to Generate Custom QR Codes with Python

QR codes are everywhere. They handle contactless restaurant menus, physical tickets, inventory tags, and quick-login screens. But while adding a static QR code is easy, generating custom, dynamic, styled QR codes programmatically on your backend can quickly turn into an operational bottleneck.

Using the QR Code Generator on Apify, you can generate perfectly styled, high-resolution QR codes with a single API call. Here is how to build a dynamic QR code generator pipeline using Python.

The Messy State of Server-Side QR Code Generation

Most Python developers start by looking up a native QR library. The standard path is to pip-install qrcode and Pillow. While this works fine on a local development machine, moving it to production brings several immediate challenges.

Pillow is a heavy graphics library with native C dependencies. If you deploy your application to AWS Lambda, Google Cloud Functions, or an Alpine Linux Docker container, compiling these dependencies often fails or requires installing extra system packages.

Furthermore, styling the QR codes — adding custom brand colors, adjusting margins, or exporting high-quality SVG formats instead of standard PNGs — requires writing complex graphics manipulation code. If your backend needs to serve hundreds of dynamic codes concurrently, running the generation code on your main server thread blocks event loops and degrades performance.

Delegating this work to a dedicated microservice API eliminates these headaches. Your server stays lightweight, you don't need any complex system dependencies, and you can focus entirely on your core application logic.

Programmatic QR Code Generation

The QR Code Generator takes parameters such as the raw text, size, color, background color, margin size, image format (PNG, SVG, or JPEG), and error correction level, and returns the generated QR image directly or uploads it to a dataset.

Here is how to request and save a custom QR code using Python and the requests library.

import requests

API_TOKEN = "YOUR_APIFY_TOKEN"
ACTOR_ID = "weeknds~qr-code-generator"

def generate_qr_code(text, filename="qr_code.png", color="#000000", bg_color="#ffffff", size=300, format="png"):
    url = f"https://api.apify.com/v2/acts/{ACTOR_ID}/run-sync-get-dataset-items?token={API_TOKEN}"

    # Configure the payload with custom styles
    payload = {
        "text": text,
        "size": size,
        "color": color,
        "backgroundColor": bg_color,
        "format": format,
        "margin": 4,
        "errorCorrectionLevel": "H" # High error correction allows damage/overlay
    }

    # Run the actor synchronously
    response = requests.post(url, json=payload)
    response.raise_for_status()
    results = response.json()

    if not results:
        raise Exception("Failed to generate QR code")

    # The actor returns the image in base64 format or as a direct file link
    qr_data = results[0]
    image_url = qr_data.get("fileUrl")

    # Download the generated image file
    img_response = requests.get(image_url)
    img_response.raise_for_status()

    with open(filename, "wb") as f:
        f.write(img_response.content)

    print(f"Successfully saved custom QR code to {filename}")

# Generate a brand-colored QR code
generate_qr_code(
    text="https://apify.com",
    filename="apify_qr.png",
    color="#FF5733",  # Custom orange/red hex color
    bg_color="#FFFFFF",
    size=400
)

By requesting the output synchronously, you can fetch, save, and serve the QR code file in under two seconds.

Dynamic Ticket Generation and Delivery

For event platforms or ticketing engines, each attendee needs a unique QR code containing their reservation ID. This code must be generated dynamically on booking and sent in a confirmation email.

Here is how to structure a reservation workflow that generates a styled QR code for a new booking:

import json
import uuid

def process_ticket_booking(user_email, event_name, ticket_type):
    # Create a unique reservation ID
    reservation_id = str(uuid.uuid4())

    # Encode structured ticket details in the QR code
    ticket_payload = {
        "reservation_id": reservation_id,
        "event": event_name,
        "type": ticket_type,
        "email": user_email
    }
    qr_payload_string = json.dumps(ticket_payload)

    # File naming convention
    filename = f"tickets/ticket_{reservation_id}.png"

    # Generate the styled QR code matching event branding
    # Using high error correction so it scans even if folded or slightly scuffed
    try:
        generate_qr_code(
            text=qr_payload_string,
            filename=filename,
            color="#2E4053",  # Deep slate gray
            bg_color="#FADBD8",  # Soft pastel pink
            size=350
        )
        return {
            "status": "success",
            "reservation_id": reservation_id,
            "ticket_file": filename,
            "message": "Ticket QR generated and saved."
        }
    except Exception as e:
        return {
            "status": "error",
            "reason": str(e)
        }

This ensures that every ticket issued is uniquely trackable, visually unique, and instantly ready for check-in scanners.

Batch Inventory QR Code Creator

If you are setting up an inventory tracking system, you likely have a CSV sheet containing hundreds of asset IDs. You need to convert all of them into physical labels.

Running these lookups in a loop is simple, but we should handle API rate limits and organize the outputs efficiently.

import csv
import time
import os

def generate_inventory_labels(csv_filepath, output_dir="labels"):
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)

    with open(csv_filepath, mode="r", encoding="utf-8") as file:
        reader = csv.DictReader(file)

        for idx, row in enumerate(reader):
            asset_id = row.get("asset_id")
            asset_name = row.get("name")

            if not asset_id:
                continue

            # Create descriptive filenames
            clean_name = asset_name.lower().replace(" ", "_")
            filename = os.path.join(output_dir, f"{asset_id}_{clean_name}.png")

            print(f"Processing label [{idx + 1}]: {asset_name} ({asset_id})...")

            try:
                generate_qr_code(
                    text=f"https://inventory.internal/assets/{asset_id}",
                    filename=filename,
                    color="#111111",
                    bg_color="#FFFFFF",
                    size=250
                )
                # Polite delay to respect API rate limits
                time.sleep(0.5)
            except Exception as e:
                print(f"Error processing asset {asset_id}: {e}")
                continue

# Mock CSV structure
# asset_id,name
# AST-992,MacBook Pro M3
# AST-993,Office Chair Herman Miller
# AST-994,Dell UltraSharp 27 Monitor

This allows you to quickly export a folder of ready-to-print labels that can be applied to physical office assets or stock warehouse boxes.

Pricing

The QR Code Generator costs $0.001 per run. Generating 10,000 unique ticket or asset labels costs exactly $10. With no base server subscription costs or infrastructure maintenance, you only pay for the exact volume of QR codes your platform generates.

Integrating the API

Generating custom, styled graphics does not require building heavy infrastructure or managing system library complications. By calling the QR generation endpoint directly, you can programmatically produce styled vector or raster files using minimal code.

How to Geolocate Any IP Address in Python with 3 Lines of Code

Cizze R — Wed, 15 Jul 2026 05:42:21 +0000

How to Geolocate Any IP Address in Python with 3 Lines of Code

If you need to know where your users are coming from — whether to show the right currency, pre-fill country selectors, or flag suspicious logins — you need an IP geolocation service. While setting this up might sound like a weekend project involving bulky database downloads and licensing agreements, you can actually solve it in minutes with a simple API call.

Using the IP Geolocation Lookup on Apify, you get sub-second lookups without the overhead of maintaining database files or paying massive monthly minimums to commercial vendors. Here is how to build a production-grade geolocation pipeline in Python.

The Geo-IP Database Headache

Most developers start by downloading a free MaxMind GeoLite2 database. It seems like the right approach until you hit the operational realities.

First, the file is large — often hundreds of megabytes — which bloats your Docker images or serverless function deployment sizes. Second, IP distributions change constantly; if you do not write a cron job to download updates every week, your location data quickly degrades. Finally, the licensing is restrictive, requiring attribution and strict compliance checks that complicate commercial apps.

An API-based solution solves this instantly. By delegating lookups to an external service, your application stays lean, database updates are handled transparently, and you only pay for what you actually query.

Quick Start with Python

The IP Geolocation Lookup takes an IP address (or a list of IPs for batch processing) and returns structured JSON with the country, region, city, postal code, latitude, longitude, timezone, internet service provider (ISP), and autonomous system number (ASN).

Here is how to run a synchronous lookup in Python using the standard requests library.

import requests

API_TOKEN = "YOUR_APIFY_TOKEN"
ACTOR_ID = "weeknds~ip-geolocation-lookup"

def geolocate_ip(ip_address):
    url = f"https://api.apify.com/v2/acts/{ACTOR_ID}/runs?token={API_TOKEN}"

    # We pass the target IP in the input payload
    payload = {
        "ip": ip_address
    }

    # Start the actor run
    response = requests.post(url, json=payload)
    response.raise_for_status()
    run_data = response.json()
    run_id = run_data["data"]["id"]

    # In a real app, wait a brief moment or use sync endpoints
    # For sub-second response, the API allows sync runs
    sync_url = f"https://api.apify.com/v2/acts/{ACTOR_ID}/run-sync-get-dataset-items?token={API_TOKEN}"
    sync_response = requests.post(sync_url, json=payload)
    sync_response.raise_for_status()

    results = sync_response.json()
    return results[0] if results else None

# Example usage
geo_data = geolocate_ip("8.8.8.8")
if geo_data:
    print(f"Country: {geo_data.get('country')}")
    print(f"City: {geo_data.get('city')}")
    print(f"ISP: {geo_data.get('isp')}")

The output JSON contains everything you need to map the user's location:

{
  "ip": "8.8.8.8",
  "country": "United States",
  "countryCode": "US",
  "region": "California",
  "regionCode": "CA",
  "city": "Mountain View",
  "zip": "94043",
  "lat": 37.4223,
  "lon": -122.0847,
  "timezone": "America/Los_Angeles",
  "isp": "Google LLC",
  "org": "Google Public DNS",
  "as": "AS15169 Google LLC"
}

Detecting Fast-Travel Login Fraud

One of the most practical security features you can build with IP geolocation is a "fast-travel" detector. If a user logs into your system from New York, and then logs in ten minutes later from London, it is physically impossible for them to have traveled that distance. One of those sessions is fraudulent.

We can calculate the physical distance between two IP addresses using the Haversine formula on their latitudes and longitudes, then divide by the time elapsed to check if the travel speed is realistic.

import math
from datetime import datetime

def haversine_distance(lat1, lon1, lat2, lon2):
    # Radius of the Earth in kilometers
    R = 6371.0

    # Convert latitude and longitude to radians
    phi1 = math.radians(lat1)
    phi2 = math.radians(lat2)
    delta_phi = math.radians(lat2 - lat1)
    delta_lambda = math.radians(lon2 - lon1)

    a = (math.sin(delta_phi / 2.0) ** 2 +
         math.cos(phi1) * math.cos(phi2) *
         math.sin(delta_lambda / 2.0) ** 2)

    c = 2.0 * math.atan2(math.sqrt(a), math.sqrt(1.0 - a))
    distance = R * c
    return distance

def analyze_login_security(user_id, current_ip, last_login_ip, last_login_time):
    # Geolocate both IPs
    current_geo = geolocate_ip(current_ip)
    last_geo = geolocate_ip(last_login_ip)

    if not current_geo or not last_geo:
        return {"action": "allow", "reason": "Could not verify locations"}

    # Calculate distance
    dist_km = haversine_distance(
        current_geo["lat"], current_geo["lon"],
        last_geo["lat"], last_geo["lon"]
    )

    # Calculate time difference in hours
    time_diff = datetime.now() - last_login_time
    hours_elapsed = time_diff.total_seconds() / 3600.0

    if hours_elapsed <= 0:
        hours_elapsed = 0.01  # Prevent division by zero

    # Calculate travel speed in km/h
    required_speed = dist_km / hours_elapsed

    # Commercial passenger jets fly around 900 km/h
    if required_speed > 950.0 and dist_km > 100.0:
        return {
            "action": "block",
            "reason": f"Impossible travel speed: {required_speed:.1f} km/h. "
                      f"Moved {dist_km:.1f} km in {hours_elapsed:.2f} hours.",
            "current_location": f"{current_geo['city']}, {current_geo['country']}",
            "last_location": f"{last_geo['city']}, {last_geo['country']}"
        }

    return {
        "action": "allow",
        "reason": f"Travel speed {required_speed:.1f} km/h is within safe bounds."
    }

This simple check runs entirely on your server and blocks unauthorized session hijacking without requiring complex machine learning setups.

Multi-Currency and Localization Routing

E-commerce websites often lose customers because they display pricing in the wrong currency or fail to localize language options. Instead of forcing users to navigate a heavy dropdown menu, you can use the visitor's IP address to automatically direct them to the correct localized store or pre-set their currency.

Here is a practical function that determines the currency and local tax tier based on the incoming request IP.

def get_localization_config(ip_address):
    geo = geolocate_ip(ip_address)
    if not geo:
        return {
            "currency": "USD",
            "tax_rate": 0.0,
            "region_code": "US",
            "language": "en"
        }

    country_code = geo.get("countryCode", "US").upper()

    # Map country codes to local configurations
    # This avoids complex third-party localized lookup tables
    config_map = {
        "US": {"currency": "USD", "tax_rate": 0.0, "language": "en"},
        "CA": {"currency": "CAD", "tax_rate": 0.05, "language": "en"},
        "GB": {"currency": "GBP", "tax_rate": 0.20, "language": "en"},
        "DE": {"currency": "EUR", "tax_rate": 0.19, "language": "de"},
        "FR": {"currency": "EUR", "tax_rate": 0.20, "language": "fr"},
        "SE": {"currency": "SEK", "tax_rate": 0.25, "language": "sv"},
        "JP": {"currency": "JPY", "tax_rate": 0.10, "language": "ja"}
    }

    # Fallback to Euro for other EU countries
    eu_countries = {"AT", "BE", "CY", "EE", "FI", "GR", "IE", "IT", "LV", "LT", "LU", "MT", "NL", "PT", "SK", "SI", "ES"}

    if country_code in config_map:
        return {
            "currency": config_map[country_code]["currency"],
            "tax_rate": config_map[country_code]["tax_rate"],
            "region_code": country_code,
            "language": config_map[country_code]["language"]
        }
    elif country_code in eu_countries:
        return {
            "currency": "EUR",
            "tax_rate": 0.20,  # Estimated fallback VAT
            "region_code": country_code,
            "language": "en"
        }
    else:
        # Global fallback
        return {
            "currency": "USD",
            "tax_rate": 0.0,
            "region_code": country_code,
            "language": "en"
        }

This localization config can be fed directly to your frontend template engine, ensuring that users see localized pricing on their very first page load.

Pricing

The IP Geolocation Lookup costs $0.001 per run. This means you can process 1,000 geolocation checks for exactly $1. For scaling applications, this is highly economical compared to major dedicated enterprise vendors that enforce high monthly platform fees and require annual commitments.

Integrating the API

The integration is lightweight and fits easily into any web framework. You do not need to install complex system libraries or manage local binary files. A standard HTTP post request is all it takes to keep your data current and accurate.

5 Developer Tools You Shouldn't Build Yourself

Cizze R — Sun, 12 Jul 2026 21:48:37 +0000

5 Developer Tools You Shouldn't Build Yourself

Every developer has a graveyard of "I'll build that myself" projects that never shipped. The funny thing is someone already built most of them, and they cost less than the electricity your laptop used while you were thinking about it.

Here are five ready-to-use APIs that cover common developer needs. Each one costs under a cent per call.

QR codes without the library hunt

You need QR codes. You Google "python qr code library" and find twelve options. You pick one, it works, then a colleague asks for SVG output with custom colors and error correction level H. You're back to square one.

The QR Code Generator handles it in one call:

import requests

resp = requests.post(
    "https://api.apify.com/v2/acts/weeknds~qr-code-generator/runs",
    headers={"Authorization": f"Bearer {API_TOKEN}"},
    json={
        "text": "https://apify.com/weeknds",
        "format": "svg",
        "size": 512,
        "fgColor": "#1a1a2e",
        "bgColor": "#ffffff",
        "errorCorrection": "H"
    }
)

$0.002 per run. 500 QR codes for a dollar. The qrcode Python library handles basics, but the moment you need production-grade output — SVG rendering, color space conversions, error correction — you're deep in dependency management.

IP geolocation without the rate limits

Free IP geolocation APIs exist. They also throttle you after fifty requests, demand attribution links, or go down without warning.

The IP Geolocation Lookup chains multiple free providers with automatic fallback. If ip-api.com is down or rate-limited, it falls through to ipinfo.io. You always get a response:

def geolocate(ip):
    resp = requests.post(
        "https://api.apify.com/v2/acts/weeknds~ip-geolocation-lookup/runs",
        headers={"Authorization": f"Bearer {API_TOKEN}"},
        json={"ip": ip}
    )
    return get_results(resp)  # country, city, ISP, ASN, timezone, coordinates

$0.003 per run. 333 lookups for a dollar. Free APIs cap you hard. Paid ones start at $50 a month. This chains free providers behind one stable endpoint.

Link previews without the HTML parsing

Every messaging app needs link unfurling. Open Graph tags, Twitter Cards, JSON-LD, basic meta. That means handling HTTP redirects, detecting encodings (and dealing with servers that lie about them), parsing meta tags case-insensitively, extracting nested JSON-LD graphs, and falling back to title tags when nothing else exists.

The Link Preview & Metadata Extractor handles all of it:

def unfurl_link(url):
    resp = requests.post(
        "https://api.apify.com/v2/acts/weeknds~link-preview-metadata-extractor/runs",
        headers={"Authorization": f"Bearer {API_TOKEN}"},
        json={"url": url}
    )
    meta = get_results(resp)
    return {
        "title": meta["og:title"],
        "description": meta["og:description"],
        "image": meta["og:image"]
    }

$0.002 per run. 500 extractions for a dollar.

Screenshots without Puppeteer nightmares

Puppeteer in production is a headache. Memory leaks, zombie processes, Chrome version mismatches. Five minutes in and you're writing a health check for your screenshot service instead of shipping.

The Website Screenshot API handles full-page or viewport captures:

def screenshot(url):
    resp = requests.post(
        "https://api.apify.com/v2/acts/weeknds~website-screenshot-api/runs",
        headers={"Authorization": f"Bearer {API_TOKEN}"},
        json={"url": url, "fullPage": True, "viewportWidth": 1280}
    )
    return get_results(resp)["screenshotUrl"]

$0.003 per run. 333 screenshots for a dollar. Running your own headless Chrome costs more in compute alone, before you even get to proxy rotation, CAPTCHAs, and retry logic.

Visual regression testing without the build pipeline

You deploy a CSS change. Three days later someone notices the checkout button is behind the footer on mobile. Visual regression tools like Percy or Chromatic exist, but they need CI pipeline config, YAML files, and monthly subscriptions.

The Screenshot Comparison Tool takes two screenshots and produces a pixel-level comparison:

def check_for_regression(url, baseline_date):
    baseline = get_screenshot(url, baseline_date)
    current = screenshot(url)

    resp = requests.post(
        "https://api.apify.com/v2/acts/weeknds~screenshot-comparison-tool/runs",
        headers={"Authorization": f"Bearer {API_TOKEN}"},
        json={"baselineImageUrl": baseline, "currentImageUrl": current}
    )

    diff = get_results(resp)
    if diff["diffPercentage"] > 2.0:
        alert(f"Visual regression: {diff['diffPercentage']}% changed")

$0.005 per comparison. 200 comparisons for a dollar.

The numbers

Run all five at production scale and here's what it costs at 1,000 calls each:

QR Code Generator: $2.00. IP Geolocation: $3.00. Metadata Extractor: $2.00. Screenshot API: $3.00. Screenshot Diff: $5.00.

That's five fully managed API services for $15 total at 1,000 calls apiece. Your time is worth more than that.

How to Access Swedish Company Data via API

Cizze R — Fri, 10 Jul 2026 12:00:53 +0000

How to Access Swedish Company Data via API

If you're doing business in Sweden — compliance checks, supplier verification, market research — you need Bolagsverket, the official company registry. They have a search interface, but no public API.

The Swedish Company Registry scrapes Bolagsverket's Next.js frontend and returns structured JSON. Search by company name or organization number and get back everything in one call.

What you get

{
  "org_number": "556703-7482",
  "name": "Spotify AB",
  "legal_form": "Aktiebolag",
  "status": "Registrerat",
  "address": "Regeringsgatan 19, 111 53 Stockholm",
  "county": "Stockholms län",
  "registration_date": "2006-02-23",
  "officers": [
    {"name": "Daniel Ek", "role": "Verkställande direktör"}
  ]
}

Using it

import requests, time

API_TOKEN = "YOUR_APIFY_TOKEN"
ACTOR_ID = "weeknds~swedish-company-registry"

def search_company(query):
    resp = requests.post(
        f"https://api.apify.com/v2/acts/{ACTOR_ID}/runs",
        headers={"Authorization": f"Bearer {API_TOKEN}"},
        json={"searchQuery": query, "maxResults": 10}
    )
    run_id = resp.json()["data"]["id"]
    time.sleep(10)

    items = requests.get(
        f"https://api.apify.com/v2/acts/{ACTOR_ID}/runs/{run_id}/dataset/items",
        headers={"Authorization": f"Bearer {API_TOKEN}"}
    ).json()
    return items

results = search_company("Spotify")
for company in results:
    print(f"{company['name']} ({company['org_number']})")
    print(f"  {company['address']}")

Supplier verification

Check if a supplier is actually registered:

def verify_supplier(org_number):
    results = search_company(org_number)

    if not results:
        return {"valid": False, "reason": "Not found in Bolagsverket"}

    company = results[0]

    if company["status"] != "Registrerat":
        return {"valid": False, "reason": f"Status: {company['status']}"}

    return {
        "valid": True,
        "name": company["name"],
        "org_number": company["org_number"],
        "registered_since": company["registration_date"],
        "address": company["address"]
    }

Market research

Find all companies matching a keyword and see their distribution across counties and legal forms:

def analyze_industry(keyword):
    results = search_company(keyword)

    stats = {
        "total_found": len(results),
        "by_county": {},
        "by_legal_form": {}
    }

    for company in results:
        county = company.get("county", "Unknown")
        stats["by_county"][county] = stats["by_county"].get(county, 0) + 1

        form = company.get("legal_form", "Unknown")
        stats["by_legal_form"][form] = stats["by_legal_form"].get(form, 0) + 1

    return stats

Compliance checks

Flag companies that might need a closer look — recently registered entities or ones with non-standard status:

from datetime import datetime, timedelta

def check_compliance(org_number):
    results = search_company(org_number)
    if not results:
        return {"error": "Company not found"}

    company = results[0]
    flags = []

    reg_date = datetime.strptime(company["registration_date"], "%Y-%m-%d")
    if reg_date > datetime.now() - timedelta(days=90):
        flags.append("Recently registered (< 90 days)")

    if company["status"] != "Registrerat":
        flags.append(f"Non-standard status: {company['status']}")

    return {
        "org_number": org_number,
        "name": company["name"],
        "flags": flags,
        "risk_level": "high" if len(flags) > 1 else "low" if not flags else "medium"
    }

What to know

Two inputs: searchQuery (company name or org number) and maxResults (defaults to 10).

Bolagsverket rate-limits requests. The actor handles this automatically, but bulk queries will be slower than single lookups. Search quality depends on Bolagsverket's own indexing, and data reflects whatever they currently show — usually updated within 24 hours of official changes.

$0.003 per search. Checking 100 suppliers costs $0.30. Manual lookups on Bolagsverket's site take two to three minutes each. The actor does the same work in seconds.