Script: Bulk Schema Checker

import csv
import json
import os
import requests
from bs4 import BeautifulSoup
from typing import Any, Dict, List, Tuple
HEADERS = {
    "User-Agent": "Mozilla/5.0 (compatible; BulkSchemaValidator/1.0; +https://example.com/bot)"
}
TIMEOUT = 15
def load_urls(file_path: str) -> List[str]:
    """
    Load URLs from either:
    - .txt files: one URL per line
    - .csv files: first column assumed to contain URLs, header allowed
    """
    ext = os.path.splitext(file_path)[1].lower()
    urls: List[str] = []
    if ext == ".txt":
        with open(file_path, "r", encoding="utf-8") as f:
            for line in f:
                url = line.strip()
                if url:
                    urls.append(url)
    elif ext == ".csv":
        with open(file_path, "r", encoding="utf-8-sig", newline="") as f:
            reader = csv.reader(f)
            first_row = next(reader, None)
            if not first_row:
                return urls
            first_cell = first_row[0].strip().lower() if first_row else ""
            has_header = first_cell in {"url", "urls", "link", "links"}
            if not has_header and first_row[0].strip():
                urls.append(first_row[0].strip())
            for row in reader:
                if row and row[0].strip():
                    urls.append(row[0].strip())
    else:
        raise ValueError("Unsupported file type. Use .txt or .csv")
    return urls
def fetch_html(url: str) -> Tuple[int, str]:
    response = requests.get(url, headers=HEADERS, timeout=TIMEOUT)
    return response.status_code, response.text
def extract_jsonld(html: str) -> List[str]:
    soup = BeautifulSoup(html, "html.parser")
    blocks = soup.find_all("script", attrs={"type": "application/ld+json"})
    return [block.get_text(strip=True) for block in blocks if block.get_text(strip=True)]
def normalize_entities(data: Any) -> List[Dict[str, Any]]:
    """
    Convert a JSON-LD block into a list of entity dictionaries.
    Handles:
    - single object
    - list of objects
    - object with @graph
    """
    entities: List[Dict[str, Any]] = []
    if isinstance(data, dict):
        if "@graph" in data and isinstance(data["@graph"], list):
            for item in data["@graph"]:
                if isinstance(item, dict):
                    entities.append(item)
        else:
            entities.append(data)
    elif isinstance(data, list):
        for item in data:
            if isinstance(item, dict):
                entities.append(item)
    return entities
def validate_block_and_entities(parsed: Any, entities: List[Dict[str, Any]]) -> List[str]:
    """
    Validate JSON-LD in a way that avoids false positives for @graph structures.
    Important:
    - @context is checked at the block level, not on every entity
    - @type is checked on extracted entities
    """
    issues: List[str] = []
    if isinstance(parsed, dict):
        if "@context" not in parsed:
            issues.append("missing_@context")
    elif isinstance(parsed, list):
        # For top-level arrays, at least one object should usually carry @context
        has_context = any(isinstance(item, dict) and "@context" in item for item in parsed)
        if not has_context:
            issues.append("missing_@context")
    if not entities:
        issues.append("no_entities_extracted")
        return issues
    missing_type_count = 0
    for entity in entities:
        if "@type" not in entity:
            missing_type_count += 1
    if missing_type_count == len(entities):
        issues.append("missing_@type")
    return issues
def inspect_url(url: str) -> Dict[str, Any]:
    result: Dict[str, Any] = {
        "url": url,
        "http_status": "",
        "jsonld_blocks": 0,
        "entities_found": 0,
        "schema_types": "",
        "valid_json_blocks": 0,
        "invalid_json_blocks": 0,
        "issues": ""
    }
    try:
        status_code, html = fetch_html(url)
        result["http_status"] = status_code
        blocks = extract_jsonld(html)
        result["jsonld_blocks"] = len(blocks)
        if not blocks:
            result["issues"] = "no_jsonld_found"
            return result
        all_types: List[str] = []
        all_issues: List[str] = []
        for block in blocks:
            try:
                parsed = json.loads(block)
                result["valid_json_blocks"] += 1
                entities = normalize_entities(parsed)
                result["entities_found"] += len(entities)
                block_issues = validate_block_and_entities(parsed, entities)
                all_issues.extend(block_issues)
                for entity in entities:
                    entity_type = entity.get("@type")
                    if isinstance(entity_type, list):
                        all_types.extend([str(t) for t in entity_type])
                    elif entity_type:
                        all_types.append(str(entity_type))
            except json.JSONDecodeError:
                result["invalid_json_blocks"] += 1
                all_issues.append("invalid_json")
        unique_types = sorted(set(all_types))
        unique_issues = sorted(set(all_issues))
        result["schema_types"] = ", ".join(unique_types)
        result["issues"] = ", ".join(unique_issues) if unique_issues else "none"
        return result
    except requests.RequestException as e:
        result["issues"] = f"request_error: {type(e).__name__}"
        return result
    except Exception as e:
        result["issues"] = f"unexpected_error: {type(e).__name__}"
        return result
def run_audit(input_file: str, output_file: str) -> None:
    urls = load_urls(input_file)
    results: List[Dict[str, Any]] = []
    for url in urls:
        print(f"Checking: {url}")
        results.append(inspect_url(url))
    fieldnames = [
        "url",
        "http_status",
        "jsonld_blocks",
        "entities_found",
        "schema_types",
        "valid_json_blocks",
        "invalid_json_blocks",
        "issues"
    ]
    with open(output_file, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(results)
    print(f"\nDone. Report saved to: {output_file}")
if __name__ == "__main__":
    run_audit("urls.csv", "schema_audit_report.csv")
    
        """
    Important: Change input urls.csv with your own file name (csv or txt).
    """