Toplanan Verileri Doğrulama ve Şema Sapmalarını Tespit Etme

Çıkarılan veriler nadiren göze çarpan bir şekilde hata verir. Yavaş yavaş bozulur: Bir yeniden tasarım, fiyatı farklı bir öğeye taşır; ayrıştırıcı None sonucunu döndürür ve fiyatları boş olan on bin satır, tek bir istisna bile olmadan veri ambarına yüklenir. Bu kılavuz, şema sapmasının kötü bir rapora dönüşmeden önce onu yakalayan katmanları oluşturur: bir pydantic v2 kayıt şeması, parti başına kalite kontrolleri, kayan bir referans çizgisine sahip bir sapma dedektörü, bir karantina adımı, kanarya URL’lerinde günlük seçici sözleşme testi ve sorunu düzelttikten sonra verileri geriye dönük olarak doldurabilmeniz için ayrıştırıcı sürümleme.

Çıkarılan veriler neden fark edilmeden bozulur?

Şema sapmasının üç kaynağı, neredeyse tüm sorunların kaynağıdır. Sitenin yeniden tasarlanması DOM’u değiştirir ve .get() ile find() kullanılarak yazılmış toleranslı bir ayrıştırıcı, hata sinyali vermek yerine boş sonuçlar döndürür. A/B düzenleri daha da kötüdür çünkü isteklerin sadece bir kısmı yeni şablonu görür; bu nedenle doldurma oranınız yüzde 99'dan yüzde 70'e düşer ve bir hafta boyunca kimse bunu fark etmez. Yerel ayar ve coğrafi değişiklikler, yapıyı değiştirmeden formatları değiştirir: 1,299.00, 1.299,00 olur; bir tarih, ISO biçiminden gün-önce biçimine geçer ve değer hiç ayrıştırılmamak yerine yanlış ayrıştırılır.

Ortak özellik, ayrıştırıcının çökmemesidir. Kayıt düzeyinde tolerans doğru varsayılan ayardır, ancak bu, doğruluğun başka bir yerde sağlanması gerektiği anlamına gelir. Bu yer, web kazıma boru hatlarındaki toplu iş düzeyinde veri doğrulamasıdır ve bunu oluşturmak maliyet açısından ucuzdur.

pydantic v2 ile bir kayıt şeması tanımlama

Öncelikle geçerli bir kaydın nasıl olması gerektiğini belirleyin. Zorunlu alanlar, türler, aralıklar ve biçimler bir pydantic modeline dönüşür ve ayrıştırılan her sözlük, başka bir yere gönderilmeden önce model_validate’dan geçer.

from __future__ import annotations

import re
from datetime import date, datetime
from typing import Optional

from pydantic import BaseModel, Field, ValidationError, field_validator

PRICE_RE = re.compile(r"^\d+(\.\d{1,2})?$")
SKU_RE = re.compile(r"^[A-Z0-9-]{4,32}$")

class ProductRecord(BaseModel):
    source_url: str = Field(min_length=10)
    sku: str
    title: str = Field(min_length=3, max_length=300)
    price: float = Field(gt=0, lt=100_000)
    currency: str = Field(pattern=r"^[A-Z]{3}$")
    in_stock: bool
    rating: Optional[float] = Field(default=None, ge=0, le=5)
    scraped_at: datetime
    listed_on: Optional[date] = None
    parser_version: str

    @field_validator("sku")
    @classmethod
    def sku_shape(cls, v: str) -> str:
        v = v.strip().upper()
        if not SKU_RE.match(v):
            raise ValueError(f"sku does not match pattern: {v!r}")
        return v

    @field_validator("price", mode="before")
    @classmethod
    def normalise_price(cls, v):
        # Accept "1,299.00" or "1299" from the parser; reject anything else.
        if isinstance(v, str):
            cleaned = v.replace(",", "").replace("$", "").strip()
            if not PRICE_RE.match(cleaned):
                raise ValueError(f"price not parseable: {v!r}")
            return float(cleaned)
        return v

    @field_validator("title")
    @classmethod
    def title_not_placeholder(cls, v: str) -> str:
        if v.strip().lower() in {"", "n/a", "untitled", "loading..."}:
            raise ValueError("title looks like a placeholder")
        return v.strip()

def validate_batch(rows: list[dict]) -> tuple[list[ProductRecord], list[dict]]:
    good: list[ProductRecord] = []
    bad: list[dict] = []
    for row in rows:
        try:
            good.append(ProductRecord.model_validate(row))
        except ValidationError as e:
            bad.append({"row": row, "errors": e.errors(include_url=False)})
    return good, bad

Burada iki seçim önemlidir. Fiyat doğrulayıcısındaki mode="before", şemanın son normalleştirme işlemini gerçekleştirmesini sağlar; böylece ayrıştırıcı basit kalabilir. e.errors() ise loc tuple'ı ile yapılandırılmış hatalar döndürür; bu sayede hataları kayıt başına değil, alan başına sayabilirsiniz. Satırların yüzde 40’unun sadece price’da hata verdiği ve başka hiçbir yerde hata görülmeyen bir toplu iş, her satırın her alanda hata verdiği bir toplu işten farklı bir sorundur.

Toplu iş kontrolleri

Doğrulama işlemi satırları reddeder. Toplu iş kontrolleri, testi geçen ve geçmeyen satırları bir arada değerlendirerek, toplu işin bir bütün olarak makul olup olmadığını sorgular. Dört kontrol, şema sapmalarının çoğunu tespit eder.

from collections import Counter
from statistics import median

def batch_metrics(good: list[ProductRecord], bad: list[dict], fields: list[str]) -> dict:
    total = len(good) + len(bad)
    failures: Counter[str] = Counter()
    for b in bad:
        for err in b["errors"]:
            if err["loc"]:
                failures[str(err["loc"][0])] += 1

    fill_rate = {}
    for f in fields:
        present = sum(1 for g in good if getattr(g, f, None) is not None)
        fill_rate[f] = present / total if total else 0.0

    skus = [g.sku for g in good]
    dup_rate = 1 - (len(set(skus)) / len(skus)) if skus else 0.0
    prices = [g.price for g in good]

    return {
        "record_count": total,
        "reject_rate": len(bad) / total if total else 0.0,
        "field_failures": dict(failures),
        "fill_rate": fill_rate,
        "duplicate_rate": dup_rate,
        "price_median": median(prices) if prices else None,
    }

def count_in_range(count: int, expected: int, tolerance: float = 0.20) -> bool:
    return abs(count - expected) <= expected * tolerance

Her çalıştırmada batch_metrics'yu saklayın. Bu, sapma dedektörünün ham verisidir ve aynı zamanda birisi Salı günkü rakamların neden tuhaf göründüğünü sorduğunda ilk bakacağınız şeydir.

Kayma dedektörü ve kayan referans çizgisi

Sabit eşik değerleri, kaynakların farklı olması nedeniyle işe yaramaz. Bir tesiste yüzde 60 oranında dolu bir alan normalken, başka bir tesiste bu durum bir kriz anlamına gelir. Bunun yerine, kaynağın kendi yakın geçmişiyle karşılaştırın: son N adet iyi partiden oluşan kayan bir referans çizgisi.

from dataclasses import dataclass, field

@dataclass
class DriftAlert:
    field_name: str
    baseline: float
    current: float
    drop_points: float

@dataclass
class DriftDetector:
    window: int = 14
    drop_threshold: float = 10.0  # percentage points
    critical_fields: dict[str, float] = field(default_factory=lambda: {"price": 5.0})
    history: list[dict] = field(default_factory=list)

    def baseline_fill(self, field_name: str) -> float | None:
        rates = [m["fill_rate"].get(field_name) for m in self.history[-self.window:]]
        rates = [r for r in rates if r is not None]
        return median(rates) if rates else None

    def check(self, metrics: dict, expected_count: int) -> list[DriftAlert]:
        alerts: list[DriftAlert] = []
        for name, current in metrics["fill_rate"].items():
            base = self.baseline_fill(name)
            if base is None:
                continue
            drop = (base - current) * 100
            limit = self.critical_fields.get(name, self.drop_threshold)
            if drop >= limit:
                alerts.append(DriftAlert(name, base, current, drop))
        if not count_in_range(metrics["record_count"], expected_count):
            alerts.append(DriftAlert("record_count", expected_count, metrics["record_count"], 0.0))
        if metrics["duplicate_rate"] > 0.02:
            alerts.append(DriftAlert("duplicate_rate", 0.0, metrics["duplicate_rate"], 0.0))
        return alerts

    def accept(self, metrics: dict) -> None:
        # Only batches that passed become part of the baseline.
        self.history.append(metrics)

Önemli olan satır accept'dur: sadece hatasız partiler referans çizgisine dahil edilir. Sapma gösteren bir partinin girmesine izin verirseniz, referans çizgisi bu sapmayı takip eder ve dedektör tam da yüksek ses vermesi gereken anda sessiz kalır. Aynı nedenden ötürü ortalama yerine medyanı kullanın; tek bir olağandışı gün referans çizgisini değiştirmemelidir.

Yükleme yerine karantinaya alma

Algılayıcı uyarı verdiğinde, işleme grubunu yüklemeyin, ancak atmayın da. Metrikleri ve uyarıları ekli olarak bir karantina konumuna kaydedin, uyarıyı gönderin ve kararın bir kişi ya da otomatik bir kural tarafından verilmesini sağlayın. Karar genellikle üç seçenekten biridir: yanlış alarm (partiyi yükseltin ve temel veriye kabul edin), gerçek sapma (ayrıştırıcıyı düzeltin, depolanmış ham içerikten yeniden çalıştırın) veya kaynak kesintisi (partiyi atın, temel veriyi değiştirmeyin).

import json
from pathlib import Path

def route_batch(batch_id: str, good, bad, metrics, alerts, detector: DriftDetector) -> str:
    if alerts:
        q = Path("quarantine") / batch_id
        q.mkdir(parents=True, exist_ok=True)
        (q / "metrics.json").write_text(json.dumps(metrics, default=str, indent=2))
        (q / "alerts.json").write_text(json.dumps([a.__dict__ for a in alerts], indent=2))
        (q / "rows.jsonl").write_text(
            "\n".join(g.model_dump_json() for g in good)
        )
        notify(f"batch {batch_id} quarantined: " + ", ".join(a.field_name for a in alerts))
        return "quarantined"
    load_to_warehouse(good)
    detector.accept(metrics)
    return "loaded"

def notify(message: str) -> None:
    # Wire to Slack, PagerDuty, email, or a log sink.
    print("ALERT:", message)

def load_to_warehouse(records) -> None:
    pass

"Doğrulama başarısız" uyarısı yerine, neyin değiştiğine dair uyarı verin. price fill rate 0.98 -> 0.61 on source X şeklinde bir mesaj eyleme geçirilebilir; genel bir hata mesajı ise değildir.

Canary URL’lerinde seçici sözleşme testi

Toplu kontroller, bir sapma meydana geldikten sonra size bunu bildirir. Seçici sözleşme testi ise bunu daha erken, genellikle günlük çalıştırmadan önce bildirir. Kaynak başına, beklenen değerlerini bildiğiniz sabit bir kanarya URL'leri kümesi seçin, bunları her gün POST /v1/extract aracılığıyla alın, ayrıştırıcıyı çalıştırın ve sözleşmeye dahil olan her alanın mevcut ve makul olduğunu doğrulayın. Sözleşme başarısız olduğunda, büyük çalıştırma başlamadan önce ayrıştırıcıya müdahale edilmesi gerekir.

import os
import httpx

BASE = "https://scraper.geonode.io"
HEADERS = {"X-Api-Key": os.environ["GEONODE_SCRAPER_API_KEY"]}

CANARIES = [
    {"url": "https://example.com/product/123", "expect": {"sku": "EX-123", "currency": "USD"}},
    {"url": "https://example.com/product/456", "expect": {"sku": "EX-456", "currency": "USD"}},
]

CONTRACT_FIELDS = ["sku", "title", "price", "currency", "in_stock"]

def fetch_markdown(url: str) -> str:
    r = httpx.post(
        f"{BASE}/v1/extract",
        headers=HEADERS,
        json={
            "url": url,
            "formats": ["markdown"],
            "render_js": True,
            "processing_mode": "sync",
            "proxy": {"country": "US", "type": "residential"},
        },
        timeout=90.0,
    )
    r.raise_for_status()
    return r.json()["data"]["markdown"]

def parse_product(markdown: str) -> dict:
    # Your parser lives here; it should return a plain dict for the schema.
    raise NotImplementedError

def run_contract(parser_version: str) -> list[str]:
    failures: list[str] = []
    for c in CANARIES:
        md = fetch_markdown(c["url"])
        row = parse_product(md) | {"source_url": c["url"], "parser_version": parser_version}
        try:
            rec = ProductRecord.model_validate(row)
        except ValidationError as e:
            failures.append(f"{c['url']}: {[x['loc'] for x in e.errors()]}")
            continue
        for k, v in c["expect"].items():
            if getattr(rec, k) != v:
                failures.append(f"{c['url']}: {k} expected {v!r} got {getattr(rec, k)!r}")
    return failures

if __name__ == "__main__":
    problems = run_contract(parser_version="2026.10.1")
    if problems:
        notify("selector contract failed:\n" + "\n".join(problems))
        raise SystemExit(1)

Kaynak başına beş ila on kanarya yeterlidir. Site yerelleştiriliyorsa proxy ülkesini sabitleyin; çünkü farklı bir bölgeden alınan bir kanarya, farklı bir sayfayı test eder. Önem verdiğiniz düzenleri test eden URL’leri seçin: biri stokta olan, biri stokta olmayan, biri indirimde olan, biri derecelendirilmemiş.

Ayrıştırıcının sürümünü belirleyin ve ham sayfayı kaydedin

Her satırda parser_version bulunmalı ve ayrıştırıcı mantığı her değiştiğinde sürüm de değişmelidir. Bu size iki avantaj sağlar. İlk olarak, herhangi bir metriği sürüme göre dilimleyebilir ve bir sapma uyarısının bir dağıtımla çakışıp çakışmadığını anında görebilirsiniz. İkincisi, bir ayrıştırıcıyı düzelttiğinizde hangi satırları yeniden oluşturmanız gerektiğini tam olarak bilirsiniz.

Yeniden oluşturma, yalnızca girdiyi saklamış olmanız durumunda mümkündür. Her veri alımından elde edilen ham HTML veya Markdown verilerini, sıkıştırılmış olarak, source_url ve scraped_at anahtarlarıyla depolayın. Depolama maliyeti, verileri yeniden kazıma maliyetine kıyasla çok düşüktür; üstelik sayfa yeniden değiştiği için çoğu zaman yeniden kazıma bir seçenek bile olmaz. Şema sapması doğrulandığında, iş akışı şöyledir: ayrıştırıcıyı düzeltin, sürümü yükseltin, etkilenen pencere için depolanan ham içeriği yeniden ayrıştırın, aynı şema üzerinden doğrulayın ve karantinaya alınan partilerin yerine yükleyin. Ham veri depolaması olmadan, bu boşluk kalıcı olur.

import gzip
import hashlib
from pathlib import Path

def store_raw(source_url: str, scraped_at: datetime, content: str, fmt: str = "md") -> Path:
    key = hashlib.sha256(source_url.encode()).hexdigest()[:16]
    p = Path("raw") / scraped_at.strftime("%Y/%m/%d") / f"{key}.{fmt}.gz"
    p.parent.mkdir(parents=True, exist_ok=True)
    with gzip.open(p, "wt", encoding="utf-8") as f:
        f.write(content)
    return p

Sık yapılan hatalar

Kayıt bazında doğrulama yapmak ve asla toplu iş bazında yapmamak

Bir kayıt şeması, biçim hatası olan satırları yakalar; ancak her satırın biçimi doğru olmasına rağmen fiyatların yarısının eksik olduğu bir toplu işi yakalayamaz. Doluluk oranlarını ve sayıları her zaman toplu iş düzeyinde hesaplayın.

Hatalı toplu işlerin temel çizgiye girmesine izin vermek

Hareketli temel çizgi sapma gösteren bir toplu iş içeriyorsa, bir sonraki sapma gösteren toplu iş normal görünür. Yalnızca geçmişe geçen toplu işleri kabul edin ve medyan değeri kullanın.

Ayrıştırıcıyı fazla toleranslı hale getirmek

.get("price") yerine None döndüren bir ayrıştırıcı sorun yaratmaz, ancak hata durumunda 0.0 veya boş bir dize döndüren bir ayrıştırıcı, doldurma oranı kontrolünü bozar. Eksik veriler için None döndürün ve şemanın bunu reddetmesine izin verin.

Alım koşullarını sabitlememe

Canary testleri ve üretim ortamındaki çalıştırmalar aynı görüntüleme ve proxy ayarlarını kullanmalıdır. render_js: true ve bir ABD proxy'si ile geçen bir sözleşme testi, ikisini de kullanmayan bir çalıştırma hakkında hiçbir şey kanıtlamaz.

Sonraki adım

Bu kılavuzdaki tüm veri alım işlemleri, Geonode Scraper API üzerinden gerçekleştirilir; bu katman, gerektiğinde JavaScript ile işlenmiş ve ülke düzeyinde proxy hedeflemesi içeren tek bir POST /v1/extract çağrısından temiz Markdown veya HTML döndürür. Alma katmanını istikrarlı tutmak, şema sapması araştırmalarından bir değişkeni ortadan kaldırır; çünkü değişen şey sayfadır, onu nasıl elde ettiğiniz değil. Ücretsiz kademede aylık 1.5K istek hakkı sunulur; bu, doğrulama boru hattını oluştururken her gün çeşitli kaynaklara karşı kanarya sözleşmeleri çalıştırmak için yeterlidir.

SSS

Web veri kazımında şema kayması nedir?

Şema kayması, kodunuzdan ziyade kaynak sitenin değişmesi nedeniyle, ayrıştırıcınızın ürettiği verilerin yapısında meydana gelen kademeli veya ani bir değişikliktir. Bir yeniden tasarım, fiyatı yeni bir öğeye taşır; bir A/B testi, bazı istekler için farklı bir düzen oluşturur; ya da bir yerel ayar değişikliği, sayı biçimini değiştirir. Ayrıştırıcı çalışmaya devam eder, ancak alanlar boş kalır veya hatalı olur.

Kazınan veriler neden yüksek sesle hata vermeden sessizce bozulur?

Çoğu ayrıştırıcı toleranslı olacak şekilde yazılmıştır; bu nedenle eksik bir öğe, hata atmak yerine None değerini döndürür. Bu, kayıt bazında mantıklı olsa da toplu iş bazında felakete yol açar; çünkü fiyatı boş olan on bin kayıt, hata vermeden yüklenir. Bunu ancak, genellikle haftalar sonra, bir sonraki aşamadaki raporun yanlış görünmesiyle fark edersiniz. Toplu iş düzeyinde doğrulama ve sapma kontrolleri bu boşluğu kapatır.

Pydantic ile kazınan verileri nasıl doğrularım?

Gerekli alanlar, türler, sayısal aralıklar ve düzenli ifade kısıtlamalarını içeren her kayıt türü için bir model tanımlayın, ardından ValidationError'ı yakalayan bir try bloğu içinde ayrıştırılan her sözlük için model_validate işlevini çağırın. Toplu iş genelinde alan başına hata sayısını sayın. Bir alanda yüksek reddedilme oranı, şema sapması sinyalidir; her yerde yüksek oran ise genellikle veri alımının kendisinin başarısız olduğu anlamına gelir.

Şema sapması uyarısını tetikleyecek eşik değeri ne olmalıdır?

Mevcut partideki her alanın doldurulma oranını, son zamanlardaki hatasız partilerin kayan bir referans değeriyle karşılaştırın. Zorunlu bir alanda yüzde on veya daha fazla bir düşüş, güçlü bir işarettir; fiyat gibi yüksek değerli alanlar için bu eşiği yüzde beşe ayarlayın. Ayrıca kayıt sayısını beklenen değerle karşılaştırırken artı veya eksi yüzde 20 aralığını göz önünde bulundurun ve yinelenen kayıt oranını izleyin. Bir haftalık veri toplandıktan sonra eşikleri kaynağa göre ayarlayın.

Her veri toplama işleminden elde edilen ham HTML veya Markdown verilerini saklamalı mıyım?

Evet, depolama alanı izin veriyorsa. Ham HTML veya Markdown, bir ayrıştırıcıyı düzelttikten sonra verileri geriye dönük olarak doldurmanıza olanak tanıyan tek şeydir; aksi takdirde, veri toplama anındaki sayfa kaybolmuş olur. Sıkıştırılmış HTML, yeniden veri toplama maliyetine kıyasla küçüktür ve ayrıca bir alanın neden sapma gösterdiğini teşhis etmek için ihtiyacınız olan kanıtı da sağlar.

Doğrulama, topladığım verilerin doğru olduğunu garanti edebilir mi?

Hayır. Doğrulama ve sapma tespiti, biçimi ve dağılımdaki değişiklikleri yakalar; ancak ayrıştırılan bir değerin, bir insanın sayfada göreceği ile eşleşip eşleşmediğini doğrulayamaz. Bir alan, anlamı değişse bile doldurulmuş olarak kalabilir. Otomatik kontrolleri periyodik manuel örneklemeyle birleştirin ve veri kalitesinden bağımsız olarak, veri toplama işleminin her sitenin şartlarına ve geçerli yasalara uyması gerektiğini unutmayın.