Scraper Endo

"""
scraper.py — Crawl endocrinology.dk/nbv -> extract text -> chunk -> embed (Azure OpenAI) -> store in MSSQL

Install:
pip install requests beautifulsoup4 tiktoken openai pyodbc

Environment variables:
AZURE_OPENAI_ENDPOINT=https://.openai.azure.com/
AZURE_OPENAI_API_KEY=...
AZURE_OPENAI_API_VERSION=2024-02-01 # optional
AZURE_OPENAI_EMBED_DEPLOYMENT=

MSSQL_CONN_STR=Driver={ODBC Driver 18 for SQL Server};Server=...;Database=...;Uid=...;Pwd=...;Encrypt=yes;TrustServerCertificate=no;

Notes:

• This script stores embeddings as float32 little-endian in VARBINARY(MAX).
• Retrieval can do candidate filtering (full-text/site) + cosine rerank in your backend.
  """

import os
import re
import time
import hashlib
import struct
from typing import List, Tuple, Optional
from urllib.parse import urljoin

import requests
import pyodbc
from bs4 import BeautifulSoup
import tiktoken
from openai import AzureOpenAI

----------------------------

Config

----------------------------

START = "https://endocrinology.dk/nbv/"
ALLOWED_PREFIX = START
USER_AGENT = "ClinicalGuidelineBot/1.0 (contact: you@domain.dk)"

DELAY_S = 1.2
TIMEOUT_S = 25
MAX_PAGES = 2500

CHUNK_TOKENS = 800
CHUNK_OVERLAP = 120
TIKTOKEN_ENCODING = "cl100k_base"

Azure embeddings batch size (tune to your rate limits)

EMBED_BATCH = 64

----------------------------

Helpers

----------------------------

def normalize_url(url: str) -> str:
url = (url or "").strip()
return url.split("#", 1)[0]

def is_allowed(url: str) -> bool:
return normalize_url(url).startswith(ALLOWED_PREFIX)

def sha256_text(s: str) -> str:
return hashlib.sha256(s.encode("utf-8")).hexdigest()

def floats_to_varbinary_le(floats: List[float]) -> bytes:
# float32 little-endian
return struct.pack("<" + "f" * len(floats), *floats)

def domain_from_url(url: str) -> str:
return re.sub(r"^https?://", "", url).split("/", 1)[0]

----------------------------

Scrape + extract

----------------------------

def fetch_html(session: requests.Session, url: str) -> Optional[str]:
r = session.get(url, timeout=TIMEOUT_S)
r.raise_for_status()
ct = r.headers.get("Content-Type", "")
if "text/html" not in ct:
return None
return r.text

def extract_main_text(html: str) -> str:
soup = BeautifulSoup(html, "html.parser")

# Remove noise
for tag in soup(["script", "style", "noscript", "svg"]):
    tag.decompose()

# Remove common chrome if present
for tag in soup.select("nav, footer, header"):
    tag.decompose()

main = soup.find("main") or soup.find("article") or soup.body
if not main:
    return ""

pieces = []
for el in main.find_all(["h1", "h2", "h3", "h4", "p", "li", "th", "td"]):
    txt = el.get_text(" ", strip=True)
    if not txt or len(txt) <= 2:
        continue
    pieces.append(txt)

text = "\n".join(pieces)
text = re.sub(r"[ \t]+", " ", text)
text = re.sub(r"\n{3,}", "\n\n", text).strip()
return text

def crawl_nbv() -> List[Tuple[str, str]]:
session = requests.Session()
session.headers.update({"User-Agent": USER_AGENT})

seen = set()
queue = [START]
out: List[Tuple[str, str]] = []

while queue and len(out) < MAX_PAGES:
    url = normalize_url(queue.pop(0))
    if not is_allowed(url) or url in seen:
        continue
    seen.add(url)

    try:
        html = fetch_html(session, url)
    except Exception:
        continue
    if not html:
        continue

    text = extract_main_text(html)
    if text:
        out.append((url, text))

    soup = BeautifulSoup(html, "html.parser")
    for a in soup.select("a[href]"):
        href = (a.get("href") or "").strip()
        if not href:
            continue
        nxt = normalize_url(urljoin(url, href))
        if is_allowed(nxt) and nxt not in seen:
            queue.append(nxt)

    time.sleep(DELAY_S)

return out

----------------------------

Chunking

----------------------------

def chunk_text_tokenwise(text: str) -> List[str]:
enc = tiktoken.get_encoding(TIKTOKEN_ENCODING)
tokens = enc.encode(text)
chunks: List[str] = []

i = 0
while i < len(tokens):
    j = min(i + CHUNK_TOKENS, len(tokens))
    chunk = enc.decode(tokens[i:j]).strip()
    if chunk:
        chunks.append(chunk)
    if j >= len(tokens):
        break
    i = max(0, j - CHUNK_OVERLAP)

return chunks

----------------------------

MSSQL schema + upserts

----------------------------

DDL = r"""
IF OBJECT_ID('dbo.GuidelineDocuments', 'U') IS NULL
BEGIN
CREATE TABLE dbo.GuidelineDocuments (
doc_id INT IDENTITY(1,1) PRIMARY KEY,
source_url NVARCHAR(2048) NOT NULL UNIQUE,
site NVARCHAR(256) NOT NULL,
title NVARCHAR(512) NULL,
content_hash CHAR(64) NOT NULL,
fetched_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME()
);
END;

IF OBJECT_ID('dbo.GuidelineChunks', 'U') IS NULL
BEGIN
CREATE TABLE dbo.GuidelineChunks (
chunk_id INT IDENTITY(1,1) PRIMARY KEY,
doc_id INT NOT NULL,
chunk_index INT NOT NULL,
chunk_hash CHAR(64) NOT NULL,
chunk_text NVARCHAR(MAX) NOT NULL,
embedding_deployment NVARCHAR(128) NOT NULL,
embedding VARBINARY(MAX) NOT NULL,
created_at DATETIME2 NOT NULL DEFAULT SYSUTCDATETIME(),
CONSTRAINT FK_GuidelineChunks_Doc FOREIGN KEY (doc_id) REFERENCES dbo.GuidelineDocuments(doc_id),
CONSTRAINT UQ_Doc_Chunk UNIQUE (doc_id, chunk_index)
);
END;
"""

def ensure_schema(conn: pyodbc.Connection) -> None:
cur = conn.cursor()
cur.execute(DDL)
conn.commit()

def upsert_document(conn: pyodbc.Connection, url: str, content_hash: str) -> Tuple[int, bool]:
"""
Returns (doc_id, changed)
changed=True if new or hash changed
"""
site = domain_from_url(url)
cur = conn.cursor()
cur.execute("SELECT doc_id, content_hash FROM dbo.GuidelineDocuments WHERE source_url = ?", url)
row = cur.fetchone()

if row:
    doc_id, old_hash = int(row[0]), str(row[1])
    if old_hash == content_hash:
        return doc_id, False
    cur.execute(
        "UPDATE dbo.GuidelineDocuments SET content_hash = ?, fetched_at = SYSUTCDATETIME(), site = ? WHERE doc_id = ?",
        content_hash, site, doc_id
    )
    conn.commit()
    return doc_id, True

cur.execute(
    "INSERT INTO dbo.GuidelineDocuments (source_url, site, content_hash) VALUES (?, ?, ?)",
    url, site, content_hash
)
conn.commit()
cur.execute("SELECT doc_id FROM dbo.GuidelineDocuments WHERE source_url = ?", url)
doc_id = int(cur.fetchone()[0])
return doc_id, True

def clear_chunks_for_doc(conn: pyodbc.Connection, doc_id: int) -> None:
cur = conn.cursor()
cur.execute("DELETE FROM dbo.GuidelineChunks WHERE doc_id = ?", doc_id)
conn.commit()

def upsert_chunk(
conn: pyodbc.Connection,
doc_id: int,
chunk_index: int,
chunk_text: str,
embedding_deployment: str,
embedding_bytes: bytes,
) -> None:
cur = conn.cursor()
chash = sha256_text(chunk_text)
cur.execute(
"""
MERGE dbo.GuidelineChunks AS tgt
USING (SELECT ? AS doc_id, ? AS chunk_index) AS src
ON (tgt.doc_id = src.doc_id AND tgt.chunk_index = src.chunk_index)
WHEN MATCHED THEN
UPDATE SET chunk_hash=?, chunk_text=?, embedding_deployment=?, embedding=?, created_at=SYSUTCDATETIME()
WHEN NOT MATCHED THEN
INSERT (doc_id, chunk_index, chunk_hash, chunk_text, embedding_deployment, embedding)
VALUES (?, ?, ?, ?, ?, ?);
""",
doc_id, chunk_index,
chash, chunk_text, embedding_deployment, pyodbc.Binary(embedding_bytes),
doc_id, chunk_index, chash, chunk_text, embedding_deployment, pyodbc.Binary(embedding_bytes),
)
conn.commit()

----------------------------

Azure OpenAI embeddings

----------------------------

def build_azure_client() -> AzureOpenAI:
endpoint = os.environ["AZURE_OPENAI_ENDPOINT"]
api_key = os.environ["AZURE_OPENAI_API_KEY"]
api_version = os.environ.get("AZURE_OPENAI_API_VERSION", "2024-02-01")
return AzureOpenAI(
azure_endpoint=endpoint,
api_key=api_key,
api_version=api_version,
)

def embed_texts(client: AzureOpenAI, deployment: str, texts: List[str]) -> List[List[float]]:
resp = client.embeddings.create(model=deployment, input=texts)
return [d.embedding for d in resp.data]

----------------------------

Main

----------------------------

def main():
mssql_conn_str = os.environ.get("MSSQL_CONN_STR")
if not mssql_conn_str:
raise RuntimeError("Missing MSSQL_CONN_STR env var")

embed_deployment = os.environ.get("AZURE_OPENAI_EMBED_DEPLOYMENT")
if not embed_deployment:
    raise RuntimeError("Missing AZURE_OPENAI_EMBED_DEPLOYMENT env var")

azure_client = build_azure_client()

print("Crawling NBV...")
pages = crawl_nbv()
print(f"Fetched {len(pages)} pages with text")

conn = pyodbc.connect(mssql_conn_str)
ensure_schema(conn)

updated_docs = 0
total_chunks = 0

for url, text in pages:
    doc_hash = sha256_text(text)
    doc_id, changed = upsert_document(conn, url, doc_hash)
    if not changed:
        continue

    updated_docs += 1
    chunks = chunk_text_tokenwise(text)
    if not chunks:
        continue

    # Overwrite chunks if doc changed
    clear_chunks_for_doc(conn, doc_id)

    for i in range(0, len(chunks), EMBED_BATCH):
        batch = chunks[i : i + EMBED_BATCH]
        vecs = embed_texts(azure_client, embed_deployment, batch)

        for j, (chunk_text, vec) in enumerate(zip(batch, vecs), start=i):
            emb_bytes = floats_to_varbinary_le(vec)
            upsert_chunk(
                conn=conn,
                doc_id=doc_id,
                chunk_index=j,
                chunk_text=chunk_text,
                embedding_deployment=embed_deployment,
                embedding_bytes=emb_bytes,
            )
            total_chunks += 1

    print(f"Updated: {url}  (doc_id={doc_id}, chunks={len(chunks)})")

print(f"Done. Updated docs: {updated_docs}, total chunks written: {total_chunks}")

if name == "main":
main()

The issue provides scraper.py code but no requested change, target repository file, tests, or acceptance criteria. First clarify whether this script belongs in actions/starter-workflows and what behavior is expected; then identify how it should be validated, since no test or done condition is provided.