backend/data_minning/base_scrapper.py

from dataclasses import asdict
from datetime import datetime
import io
import json
import os
import time
from typing import Any, Callable, Dict, Iterable, List, Optional, Union
from pathlib import Path
from click import Tuple
from pymongo import MongoClient, UpdateOne, errors

from .dto.stream_opts import StreamOptions

from .dto.recipe_doc import RecipeDoc

from .soup_client import SoupClient
from backend.utils.sanitization import clean
from bs4 import BeautifulSoup
from backend.config.database import db_settings

class JsonArraySink:
    """
    Append-safe JSON array writer.
    - Creates file with `[` ... `]`
    - If file exists, removes trailing `]`, appends items, and re-closes.
    """
    def __init__(self, path: str):
        self.path = path
        self._opened = False
        self._first = True
        self.f = None

    def _prepare(self):
        if self._opened:
            return

        # Ensure file exists with an empty array
        if not os.path.exists(self.path):
            with open(self.path, "w", encoding="utf-8") as f:
                f.write("[\n]")

        self.f = open(self.path, "r+", encoding="utf-8")

        # Find the position of the final ']' from the end
        self.f.seek(0, io.SEEK_END)
        end = self.f.tell()
        step = min(4096, end)
        pos = end
        last_bracket = -1
        while pos > 0:
            pos = max(0, pos - step)
            self.f.seek(pos)
            chunk = self.f.read(step)
            j = chunk.rfind("]")
            if j != -1:
                last_bracket = pos + j
                break

        if last_bracket == -1:
            # Corrupt file: reset to empty array
            self.f.seek(0); self.f.truncate(0); self.f.write("[\n]"); self.f.flush()
            last_bracket = 2  # index of ']' in "[\n]"

        # Decide "is first item?" by inspecting the content BEFORE the ']'
        self.f.seek(0)
        prefix = self.f.read(last_bracket).strip()   # content up to (but not including) ']'
        # Empty array has only '[' (possibly with whitespace/newline)
        self._first = (prefix == "[")

        # Now remove the closing ']' so we can append
        self.f.seek(last_bracket)
        self.f.truncate()

        self._opened = True

    def write_many(self, docs: List[Dict[str, Any]]):
        if not docs:
            return
        self._prepare()

        for d in docs:
            if not self._first:
                self.f.write(",\n")
            else:
                # First item: no leading comma
                self._first = False
            self.f.write(json.dumps(d, ensure_ascii=False, indent=2, default=str))

        # Restore the closing bracket
        self.f.write("\n]")
        self.f.flush()

    def close(self):
        if self.f:
            self.f.close()
        self._opened = False

class MongoSink:
    def __init__(self, ):
        db_config = db_settings.get_vector_store_config()
        self.client = MongoClient(db_config["uri"], retryWrites=True, serverSelectionTimeoutMS=10000)
        self.col = self.client[db_config["database"]][db_config["collection_name"]]
        self._ensure_indexes()

    def _ensure_indexes(self):
        self.col.create_index("url", unique=True)
        self.col.create_index("title")
        self.col.create_index("category")
        self.col.create_index("scraped_at")

    def write_many(self, docs: List[Dict[str, Any]]):
        if not docs: return
        ops = []
        now = datetime.utcnow()
        for d in docs:
            d = d.copy()
            d.setdefault("scraped_at", now)
            ops.append(UpdateOne({"url": d["url"]}, {"$set": d, "$setOnInsert": {"created_at": now}}, upsert=True))
        try:
            self.col.bulk_write(ops, ordered=False)
        except errors.BulkWriteError as e:
            # duplicates or minor issues won't halt unordered bulk
            pass

    def close(self):
        self.client.close()

    # we need to create a search function to fetch recipes by title or ingredients from the embeddings given the embedding fields, db and collection 

class DualSink:
    def __init__(self, json_sink: Optional[JsonArraySink], mongo_sink: Optional[MongoSink]):
        self.json = json_sink
        self.mongo = mongo_sink
    def write_many(self, docs: List[Dict[str, Any]]):
        if self.json: self.json.write_many(docs)
        if self.mongo: self.mongo.upsert_batch(docs)
    def close(self):
        if self.json: self.json.close()
        if self.mongo: self.mongo.close()


class BaseRecipeScraper(SoupClient):
    HEADING_TAGS = ("h1","h2","h3","h4","h5","h6")
    
    def __init__(
        self,
        *args,
        embedder= None,
        embedding_fields= None,
        **kwargs
    ):
        """
        embedder: HFEmbedder(), optional
        embedding_fields: list of (source_field, target_field) like:
            [("title", "title_emb"), ("instructions_text", "instr_emb")]
        """
        super().__init__(*args, **kwargs)
        self.embedder = embedder
        self.embedding_fields = embedding_fields or []
        self.logger = self.log

    def extract_jsonld(self, soup: BeautifulSoup) -> Optional[Dict[str, Any]]:
        def to_list(x): return x if isinstance(x, list) else [x]
        for tag in soup.find_all("script", type="application/ld+json"):
            try:
                data = json.loads(tag.string or "{}")
            except Exception:
                continue
            nodes = (data.get("@graph", [data]) if isinstance(data, dict)
                     else (data if isinstance(data, list) else []))
            for n in nodes:
                if not isinstance(n, dict): continue
                t = n.get("@type")
                if t == "Recipe" or (isinstance(t, list) and "Recipe" in t):
                    doc = RecipeDoc()
                    doc.title = clean(n.get("name"))
                    # ingredients
                    ings = []
                    for ing in to_list(n.get("recipeIngredient") or []):
                        if isinstance(ing, dict):
                            ings.append(clean(ing.get("name") or ing.get("text")))
                        else:
                            ings.append(clean(str(ing)))
                    doc.ingredients = [x for x in ings if x]
                    # instructions
                    steps = []
                    for st in to_list(n.get("recipeInstructions") or []):
                        if isinstance(st, dict):
                            steps.append(clean(st.get("text") or st.get("name")))
                        else:
                            steps.append(clean(str(st)))
                    doc.instructions = [x for x in steps if x]
                   
                    doc.servings = n.get("recipeYield")
                    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")))
                    doc.course = clean(n.get("recipeCategory")) if isinstance(n.get("recipeCategory"), str) else None
                    doc.cuisine = clean(n.get("recipeCuisine")) if isinstance(n.get("recipeCuisine"), str) else None
                    return asdict(doc)
        return None
    
    @staticmethod
    def _dedupe_preserve_order(items: List[str]) -> List[str]:
        seen = set()
        out = []
        for x in items:
            x = clean(x)
            if not x or x in seen: 
                continue
            seen.add(x); out.append(x)
        return out

    @staticmethod
    def _to_ingredients_text(items: List[str]) -> str:
        """
        Turn ingredient bullets into a single text block.
        Using one-per-line is great for embeddings and human readability.
        """
        items = [clean(x) for x in items if x]
        items = BaseRecipeScraper._dedupe_preserve_order(items)
        return "\n".join(f"- {x}" for x in items)

    @staticmethod
    def _to_instructions_text(steps: List[str]) -> str:
        """
        Turn ordered steps into a single text block.
        Numbered paragraphs help embeddings keep sequence context.
        """
        steps = [clean(x) for x in steps if x]
        steps = BaseRecipeScraper._dedupe_preserve_order(steps)
        return "\n\n".join(f"{i}. {s}" for i, s in enumerate(steps, 1))
    # site-specific scrapers override these two:
    def discover_urls(self) -> Iterable[str]:
        raise NotImplementedError
    def extract_recipe(self, soup: BeautifulSoup, url: str, category: Optional[str] = None) -> RecipeDoc:
        raise NotImplementedError

    # shared streaming loop
    def stream(self, sink: DualSink, options: Optional[StreamOptions] = None) -> int:
        opts = options or StreamOptions()
        self.log.info(
            f"Starting stream: limit={opts.limit} batch_size={opts.batch_size} "
            f"resume_file={opts.resume_file} sink={type(sink).__name__}"
        )

        processed = set()
        if opts.resume_file:
            resume_path = Path("data") / opts.resume_file  # <-- not ../data
            print(resume_path, 'resume_path')
            if resume_path.exists():                       # <-- open only if it exists
                with resume_path.open("r", encoding="utf-8") as f:
                    processed = {line.strip() for line in f if line.strip()}
            else:
                processed = set()
            self.log.info(f"[resume] {len(processed)} URLs already done")

        batch, saved = [], 0
        try:
            for i, url in enumerate(self.discover_urls(), 1):
                if opts.limit and i > opts.limit: break
                if not self.same_domain(url): continue
                if url in processed: continue

                try:
                    soup = self.fetch_soup(url)
                    doc = self.extract_recipe(soup, url)
                    doc.finalize()
                    batch.append(asdict(doc))
                except Exception as e:
                    self.log.warning(f"[skip] {url} -> {e}")

                if opts.resume_file:
                    resume_path = Path("data") / opts.resume_file
                    with open(resume_path, "a", encoding="utf-8") as rf:
                        rf.write(url + "\n")

                if len(batch) >= opts.batch_size:
                    self._apply_embeddings(batch)
                    sink.write_many(batch); saved += len(batch); batch = []
                    if opts.progress_callback: opts.progress_callback(saved)
                    self.log.info(f"[resume] {saved} URLs already done 1")

                if i % 25 == 0:
                    self.log.info(f"…processed {i}, saved {saved}")

                time.sleep(opts.delay)

            if batch:
                self._apply_embeddings(batch)
                sink.write_many(batch); saved += len(batch)
                if opts.progress_callback: opts.progress_callback(saved)
                self.log.info(f"[resume] {saved} URLs already done2 ")
        finally:
            sink.close()

        self.log.info(f"[done] saved {saved}")
        return saved
    
    @staticmethod
    def _field_to_text(val: Any) -> str:
        if isinstance(val, list):
            return "\n".join(str(x) for x in val)
        if val is None:
            return ""
        return str(val)

    def _gather_text(self, doc: Dict[str, Any], src: Any) -> str:
        if isinstance(src, tuple):
            parts: List[str] = []
            for f in src:
                t = self._field_to_text(doc.get(f))
                if t:
                    # Optional: label sections to help the embedder
                    label = "Ingredients" if f == "ingredients" else ("Instructions" if f == "instructions" else f)
                    parts.append(f"{label}:\n{t}")
            return "\n\n".join(parts)
        else:
            return self._field_to_text(doc.get(src))
        
    def _apply_embeddings(self, batch: List[Dict[str, Any]]) -> None:
        """
        Applies embeddings to specified fields in a batch of documents.

        For each (source_field, destination_field) pair in `self.embedding_fields`, this method:
        - Extracts the value from `source_field` in each document of the batch.
        - Converts the value to a string. If the value is a list, joins its elements with newlines.
        - Handles `None` values by converting them to empty strings.
        - Uses `self.embedder.encode` to generate embeddings for the processed texts.
        - Stores the resulting embedding vector in `destination_field` of each document.

        If `self.embedder`, `self.embedding_fields`, or `batch` is not set or empty, the method returns immediately.

        Args:
            batch (List[Dict[str, Any]]): A list of documents to process, where each document is a dictionary.

        Returns:
            None
        """
        if not self.embedder or not self.embedding_fields or not batch:
            return
        try:
            for src_spec, dst_field in self.embedding_fields:
                texts = [ self._gather_text(doc, src_spec) for doc in batch ]
                embs = self.embedder.encode(texts)
                for document, vec in zip(batch, embs):
                    document[dst_field] = vec
        except Exception as e:
            self.logger.warning(f"[stream error]: {e}")