jessejohnson/plg4-dev-server

1from dataclasses import asdict2from datetime import datetime3import io4import json5import os6import time7from typing import Any, Callable, Dict, Iterable, List, Optional, Union8from pathlib import Path9from click import Tuple10from pymongo import MongoClient, UpdateOne, errors11 12from .dto.stream_opts import StreamOptions13 14from .dto.recipe_doc import RecipeDoc15 16from .soup_client import SoupClient17from backend.utils.sanitization import clean18from bs4 import BeautifulSoup19from backend.config.database import db_settings20 21class JsonArraySink:22    """23    Append-safe JSON array writer.24    - Creates file with `[` ... `]`25    - If file exists, removes trailing `]`, appends items, and re-closes.26    """27    def __init__(self, path: str):28        self.path = path29        self._opened = False30        self._first = True31        self.f = None32 33    def _prepare(self):34        if self._opened:35            return36 37        # Ensure file exists with an empty array38        if not os.path.exists(self.path):39            with open(self.path, "w", encoding="utf-8") as f:40                f.write("[\n]")41 42        self.f = open(self.path, "r+", encoding="utf-8")43 44        # Find the position of the final ']' from the end45        self.f.seek(0, io.SEEK_END)46        end = self.f.tell()47        step = min(4096, end)48        pos = end49        last_bracket = -150        while pos > 0:51            pos = max(0, pos - step)52            self.f.seek(pos)53            chunk = self.f.read(step)54            j = chunk.rfind("]")55            if j != -1:56                last_bracket = pos + j57                break58 59        if last_bracket == -1:60            # Corrupt file: reset to empty array61            self.f.seek(0); self.f.truncate(0); self.f.write("[\n]"); self.f.flush()62            last_bracket = 2  # index of ']' in "[\n]"63 64        # Decide "is first item?" by inspecting the content BEFORE the ']'65        self.f.seek(0)66        prefix = self.f.read(last_bracket).strip()   # content up to (but not including) ']'67        # Empty array has only '[' (possibly with whitespace/newline)68        self._first = (prefix == "[")69 70        # Now remove the closing ']' so we can append71        self.f.seek(last_bracket)72        self.f.truncate()73 74        self._opened = True75 76    def write_many(self, docs: List[Dict[str, Any]]):77        if not docs:78            return79        self._prepare()80 81        for d in docs:82            if not self._first:83                self.f.write(",\n")84            else:85                # First item: no leading comma86                self._first = False87            self.f.write(json.dumps(d, ensure_ascii=False, indent=2, default=str))88 89        # Restore the closing bracket90        self.f.write("\n]")91        self.f.flush()92 93    def close(self):94        if self.f:95            self.f.close()96        self._opened = False97 98class MongoSink:99    def __init__(self, ):100        db_config = db_settings.get_vector_store_config()101        self.client = MongoClient(db_config["uri"], retryWrites=True, serverSelectionTimeoutMS=10000)102        self.col = self.client[db_config["database"]][db_config["collection_name"]]103        self._ensure_indexes()104 105    def _ensure_indexes(self):106        self.col.create_index("url", unique=True)107        self.col.create_index("title")108        self.col.create_index("category")109        self.col.create_index("scraped_at")110 111    def write_many(self, docs: List[Dict[str, Any]]):112        if not docs: return113        ops = []114        now = datetime.utcnow()115        for d in docs:116            d = d.copy()117            d.setdefault("scraped_at", now)118            ops.append(UpdateOne({"url": d["url"]}, {"$set": d, "$setOnInsert": {"created_at": now}}, upsert=True))119        try:120            self.col.bulk_write(ops, ordered=False)121        except errors.BulkWriteError as e:122            # duplicates or minor issues won't halt unordered bulk123            pass124 125    def close(self):126        self.client.close()127 128    # we need to create a search function to fetch recipes by title or ingredients from the embeddings given the embedding fields, db and collection 129 130class DualSink:131    def __init__(self, json_sink: Optional[JsonArraySink], mongo_sink: Optional[MongoSink]):132        self.json = json_sink133        self.mongo = mongo_sink134    def write_many(self, docs: List[Dict[str, Any]]):135        if self.json: self.json.write_many(docs)136        if self.mongo: self.mongo.upsert_batch(docs)137    def close(self):138        if self.json: self.json.close()139        if self.mongo: self.mongo.close()140 141 142class BaseRecipeScraper(SoupClient):143    HEADING_TAGS = ("h1","h2","h3","h4","h5","h6")144    145    def __init__(146        self,147        *args,148        embedder= None,149        embedding_fields= None,150        **kwargs151    ):152        """153        embedder: HFEmbedder(), optional154        embedding_fields: list of (source_field, target_field) like:155            [("title", "title_emb"), ("instructions_text", "instr_emb")]156        """157        super().__init__(*args, **kwargs)158        self.embedder = embedder159        self.embedding_fields = embedding_fields or []160        self.logger = self.log161 162    def extract_jsonld(self, soup: BeautifulSoup) -> Optional[Dict[str, Any]]:163        def to_list(x): return x if isinstance(x, list) else [x]164        for tag in soup.find_all("script", type="application/ld+json"):165            try:166                data = json.loads(tag.string or "{}")167            except Exception:168                continue169            nodes = (data.get("@graph", [data]) if isinstance(data, dict)170                     else (data if isinstance(data, list) else []))171            for n in nodes:172                if not isinstance(n, dict): continue173                t = n.get("@type")174                if t == "Recipe" or (isinstance(t, list) and "Recipe" in t):175                    doc = RecipeDoc()176                    doc.title = clean(n.get("name"))177                    # ingredients178                    ings = []179                    for ing in to_list(n.get("recipeIngredient") or []):180                        if isinstance(ing, dict):181                            ings.append(clean(ing.get("name") or ing.get("text")))182                        else:183                            ings.append(clean(str(ing)))184                    doc.ingredients = [x for x in ings if x]185                    # instructions186                    steps = []187                    for st in to_list(n.get("recipeInstructions") or []):188                        if isinstance(st, dict):189                            steps.append(clean(st.get("text") or st.get("name")))190                        else:191                            steps.append(clean(str(st)))192                    doc.instructions = [x for x in steps if x]193                   194                    doc.servings = n.get("recipeYield")195                    doc.image_url = clean((n.get("image") or {}).get("url") if isinstance(n.get("image"), dict) else (n.get("image")[0] if isinstance(n.get("image"), list) else n.get("image")))196                    doc.course = clean(n.get("recipeCategory")) if isinstance(n.get("recipeCategory"), str) else None197                    doc.cuisine = clean(n.get("recipeCuisine")) if isinstance(n.get("recipeCuisine"), str) else None198                    return asdict(doc)199        return None200    201    @staticmethod202    def _dedupe_preserve_order(items: List[str]) -> List[str]:203        seen = set()204        out = []205        for x in items:206            x = clean(x)207            if not x or x in seen: 208                continue209            seen.add(x); out.append(x)210        return out211 212    @staticmethod213    def _to_ingredients_text(items: List[str]) -> str:214        """215        Turn ingredient bullets into a single text block.216        Using one-per-line is great for embeddings and human readability.217        """218        items = [clean(x) for x in items if x]219        items = BaseRecipeScraper._dedupe_preserve_order(items)220        return "\n".join(f"- {x}" for x in items)221 222    @staticmethod223    def _to_instructions_text(steps: List[str]) -> str:224        """225        Turn ordered steps into a single text block.226        Numbered paragraphs help embeddings keep sequence context.227        """228        steps = [clean(x) for x in steps if x]229        steps = BaseRecipeScraper._dedupe_preserve_order(steps)230        return "\n\n".join(f"{i}. {s}" for i, s in enumerate(steps, 1))231    # site-specific scrapers override these two:232    def discover_urls(self) -> Iterable[str]:233        raise NotImplementedError234    def extract_recipe(self, soup: BeautifulSoup, url: str, category: Optional[str] = None) -> RecipeDoc:235        raise NotImplementedError236 237    # shared streaming loop238    def stream(self, sink: DualSink, options: Optional[StreamOptions] = None) -> int:239        opts = options or StreamOptions()240        self.log.info(241            f"Starting stream: limit={opts.limit} batch_size={opts.batch_size} "242            f"resume_file={opts.resume_file} sink={type(sink).__name__}"243        )244 245        processed = set()246        if opts.resume_file:247            resume_path = Path("data") / opts.resume_file  # <-- not ../data248            print(resume_path, 'resume_path')249            if resume_path.exists():                       # <-- open only if it exists250                with resume_path.open("r", encoding="utf-8") as f:251                    processed = {line.strip() for line in f if line.strip()}252            else:253                processed = set()254            self.log.info(f"[resume] {len(processed)} URLs already done")255 256        batch, saved = [], 0257        try:258            for i, url in enumerate(self.discover_urls(), 1):259                if opts.limit and i > opts.limit: break260                if not self.same_domain(url): continue261                if url in processed: continue262 263                try:264                    soup = self.fetch_soup(url)265                    doc = self.extract_recipe(soup, url)266                    doc.finalize()267                    batch.append(asdict(doc))268                except Exception as e:269                    self.log.warning(f"[skip] {url} -> {e}")270 271                if opts.resume_file:272                    resume_path = Path("data") / opts.resume_file273                    with open(resume_path, "a", encoding="utf-8") as rf:274                        rf.write(url + "\n")275 276                if len(batch) >= opts.batch_size:277                    self._apply_embeddings(batch)278                    sink.write_many(batch); saved += len(batch); batch = []279                    if opts.progress_callback: opts.progress_callback(saved)280                    self.log.info(f"[resume] {saved} URLs already done 1")281 282                if i % 25 == 0:283                    self.log.info(f"…processed {i}, saved {saved}")284 285                time.sleep(opts.delay)286 287            if batch:288                self._apply_embeddings(batch)289                sink.write_many(batch); saved += len(batch)290                if opts.progress_callback: opts.progress_callback(saved)291                self.log.info(f"[resume] {saved} URLs already done2 ")292        finally:293            sink.close()294 295        self.log.info(f"[done] saved {saved}")296        return saved297    298    @staticmethod299    def _field_to_text(val: Any) -> str:300        if isinstance(val, list):301            return "\n".join(str(x) for x in val)302        if val is None:303            return ""304        return str(val)305 306    def _gather_text(self, doc: Dict[str, Any], src: Any) -> str:307        if isinstance(src, tuple):308            parts: List[str] = []309            for f in src:310                t = self._field_to_text(doc.get(f))311                if t:312                    # Optional: label sections to help the embedder313                    label = "Ingredients" if f == "ingredients" else ("Instructions" if f == "instructions" else f)314                    parts.append(f"{label}:\n{t}")315            return "\n\n".join(parts)316        else:317            return self._field_to_text(doc.get(src))318        319    def _apply_embeddings(self, batch: List[Dict[str, Any]]) -> None:320        """321        Applies embeddings to specified fields in a batch of documents.322 323        For each (source_field, destination_field) pair in `self.embedding_fields`, this method:324        - Extracts the value from `source_field` in each document of the batch.325        - Converts the value to a string. If the value is a list, joins its elements with newlines.326        - Handles `None` values by converting them to empty strings.327        - Uses `self.embedder.encode` to generate embeddings for the processed texts.328        - Stores the resulting embedding vector in `destination_field` of each document.329 330        If `self.embedder`, `self.embedding_fields`, or `batch` is not set or empty, the method returns immediately.331 332        Args:333            batch (List[Dict[str, Any]]): A list of documents to process, where each document is a dictionary.334 335        Returns:336            None337        """338        if not self.embedder or not self.embedding_fields or not batch:339            return340        try:341            for src_spec, dst_field in self.embedding_fields:342                texts = [ self._gather_text(doc, src_spec) for doc in batch ]343                embs = self.embedder.encode(texts)344                for document, vec in zip(batch, embs):345                    document[dst_field] = vec346        except Exception as e:347            self.logger.warning(f"[stream error]: {e}")348 349