import asyncio import logging from collections import defaultdict from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import Dict, List, Optional from sqlalchemy.orm import Session from src.apps.match.enums import MatchedType, Status from src.apps.match.schemas.match_input import MatchInput from src.apps.match.schemas.match_output import MatchOutput from src.apps.match.services.keywords_service import KeywordsService from src.apps.match.services.logs_service import LogsService from src.apps.match.services.output_service import OutputService from src.apps.match.services.web_crawler import get_web_crawler from src.apps.match.services.wine_service import WineService from src.apps.match.utils.calculator import Calculator from src.apps.web_crawler.models.web_crawler import WebCrawler logger = logging.getLogger(__name__) class MatcherService: def __init__( self, output_service: OutputService = OutputService(), logs_service: LogsService = LogsService(), keywords_service: KeywordsService = KeywordsService(), wine_service: WineService = WineService(), output_encoding: str = "utf-8", default_bottle_size: str = "750ml", max_threads: int = 4, ): self.output_service = output_service self.logs_service = logs_service self.keywords_service = keywords_service self.wine_service = wine_service self.output_encoding = output_encoding self.default_bottle_size = default_bottle_size self.max_threads = max_threads self.running_keyword_matches = set() self.running_history_matches = set() def add_to_running_matches(self, retailer_code: str, match_type: MatchedType): if not retailer_code: return if match_type == MatchedType.KEYWORD: self.running_keyword_matches.add(retailer_code) elif match_type == MatchedType.HISTORY: self.running_history_matches.add(retailer_code) def remove_from_running_matches(self, retailer_code: str, match_type: MatchedType): if not retailer_code: return if match_type == MatchedType.KEYWORD: self.running_keyword_matches.discard(retailer_code) elif match_type == MatchedType.HISTORY: self.running_history_matches.discard(retailer_code) def is_terminated_match(self, retailer_code: str, match_type: MatchedType) -> bool: if not retailer_code: return True if match_type == MatchedType.KEYWORD: return retailer_code not in self.running_keyword_matches else: return retailer_code not in self.running_history_matches def clear_running_matches(self, match_type: MatchedType): if match_type == MatchedType.KEYWORD: self.running_keyword_matches.clear() else: self.running_history_matches.clear() async def execute(self, db: Session, retailer_code: str, run_keyword: bool, date: datetime): # This assumes you have WebCrawler and MatchInput equivalents web_crawler: WebCrawler = get_web_crawler(db=db, code=retailer_code) # mock this if not web_crawler: raise ValueError("Retailer not found") logger.info(f"[{retailer_code}] Matching process started") output_list = self.output_service.load_retailer_output(retailer_code, date) if not output_list: logger.warning("No output list found.") return history_matches = self.output_service.load_histories(retailer_code) results: List[MatchOutput] = [] ids = set() stats = defaultdict(int) async def process_output(output: str) -> MatchOutput: match_input = self._parse_input(web_crawler=web_crawler, output_line=output) ### match_output = MatchOutput.from_input(match_input, output, web_crawler.output_delimiter) bottle_id = match_input.bottle_size_id history_id = history_matches.get(match_output.history_string) if match_output.history_string else None if history_id: match_output.apply_history_match(history_id, self) stats["hMatch"] += 1 elif run_keyword: await self._apply_keyword_match(match_output, ids) if match_output.match == MatchedType.KEYWORD.value and not match_output.errors: stats["kMatch"] += 1 elif match_output.errors == MatchedType.AMBIGUOUS.value: stats["aMatch"] += 1 else: stats["uMatch"] += 1 return match_output tasks = [process_output(o) for o in output_list] results = await asyncio.gather(*tasks) self._set_wine_names(db, results) self.output_service.save_matched_to_file(retailer_code, results, date, run_keyword) self.logs_service.save_match_log( db, retailer_code, date, run_keyword, Status.SUCCESS, "", datetime.now(), len(output_list), stats["hMatch"], stats["kMatch"], stats["aMatch"], stats["uMatch"], ) logger.info(f"[{retailer_code}] Match completed. Stats: {dict(stats)}") async def _apply_keyword_match(self, match_output: MatchOutput, ids: set): if not match_output.vintage: match_output.match = MatchedType.UNKNOWN.value match_output.errors = "vintage missing" return match_result = await self.calculate_match(match_output.keyword_string) if match_result.matches_size == 1: wine_db_id = match_result.first_match() match_output.apply_single_keyword_match(wine_db_id, ids, self) elif match_result.matches_size > 1: match_output.apply_ambiguous_match(match_result) else: match_output.apply_unknown_match() def _parse_input(self, web_crawler, output_line: str) -> MatchInput: # Determine separator separator = getattr(getattr(web_crawler, "output_delimiter", None), "delimiter", None) if callable(separator): separator = separator() if not separator: separator = r"\|" # Split output line wine_info = output_line.split(separator) if output_line else [] # Vintage vintage = "" if getattr(web_crawler, "vintage_index", None) is not None: try: vintage = wine_info[web_crawler.vintage_index].strip() except Exception: vintage = "" if vintage and vintage.lower() == "n.v.": vintage = "NV" if not (len(vintage) == 4 or vintage.upper() == "NV"): vintage = "" # Original size info original_size_info = "" if getattr(web_crawler, "bottle_size_index", None) is not None: try: original_size_info = wine_info[web_crawler.bottle_size_index] except Exception: original_size_info = "" # Set size only if it's valid size = original_size_info if self.keywords_service.get_bottle_id(original_size_info) else "" match_input = MatchInput() try: # Price price_string = "" if getattr(web_crawler, "price_index", None) is not None: try: price_string = wine_info[web_crawler.price_index] except Exception: price_string = "" price = Calculator.parse_money(price_string) match_input.price = price # SKU sku = "" if getattr(web_crawler, "sku_index", None) is not None: try: sku = wine_info[web_crawler.sku_index] except Exception: sku = "" match_input.sku = sku # Tax Status tax_status = "" if getattr(web_crawler, "tax_status_index", None) is not None: try: tax_status = wine_info[web_crawler.tax_status_index] except Exception: tax_status = "" match_input.tax_status = tax_status # URL url = "" if getattr(web_crawler, "url_index", None) is not None: try: url = wine_info[web_crawler.url_index] except Exception: url = "" match_input.url = url # Description (space between values) desc_build = [] for it in getattr(web_crawler, "list_description_indexes", []) or []: text = wine_info[it] if it is not None and len(wine_info) > it else "" if text and text.strip(): desc_build.append(text.strip()) match_input.description = " ".join(desc_build).strip() # History input (no space between values) h_input = [] for it in getattr(web_crawler, "list_history_indexes", []) or []: text = wine_info[it] if it is not None and len(wine_info) > it else "" if text and text.strip(): h_input.append(text.strip()) h_input_str = "".join(h_input) history_input = Calculator.cleanup(h_input_str) match_input.history_text = history_input match_input.original_history_text = h_input_str.strip() # Keyword input (space between values) k_input = [] for it in getattr(web_crawler, "list_keyword_indexes", []) or []: text = wine_info[it] if it is not None and len(wine_info) > it else "" if text and text.strip(): k_input.append(text.strip()) k_input_str = " ".join(k_input) match_input.original_keyword_text = k_input_str.strip() matches_input = k_input_str matches_input = Calculator.cleanup_numeric_html(matches_input) matches_input = self.keywords_service.replace_aliases(matches_input) matches_input = self.keywords_service.remove_size_info(matches_input) # matches_input = Calculator.remove_vintages(matches_input) # commented as in Groovy matches_input = Calculator.cleanup(matches_input) matches_input = self.keywords_service.replace_aliases(matches_input) match_input.keyword_text = matches_input.strip() # Vintage parser if not vintage: vintage = Calculator.parse_vintage(match_input.original_keyword_text) if not vintage and getattr(web_crawler, "vintage_default_nv", False): vintage = "NV" match_input.vintage = vintage or "" # Bottle size parser if not size: size = self.keywords_service.parse_bottle_size(original_size_info) if not size: size = self.keywords_service.parse_bottle_size(match_input.description) if not size and not getattr(web_crawler, "bottle_size_default_blank", True): size = self.default_bottle_size match_input.size = size or "" bottle_id = self.keywords_service.get_bottle_id(size) if bottle_id: match_input.bottle_size_id = bottle_id except Exception as e: logger.error(str(e), exc_info=True) return match_input async def calculate_match(self, search_text: str): from src.apps.match.services.atomic_kw_search import AtomicKwMatch match = AtomicKwMatch() candidates = self.keywords_service.get_candidate_list(search_text) def evaluate(id_): try: score = Calculator.get_score( search_text, self.keywords_service.get_pattern(id_), self.keywords_service.get_normalized_sorted_pattern(id_), self.keywords_service.get_normalized_pattern(id_), match.score, self.keywords_service.get_required_words(id_), ) match.set(id_, score) except Exception as e: logger.error(f"Error scoring {id_}: {e}") with ThreadPoolExecutor(max_workers=self.max_threads) as executor: executor.map(evaluate, candidates) return match def _set_wine_names(self, db: Session, outputs: List[MatchOutput]): valid_outputs = [o for o in outputs if o.match == MatchedType.KEYWORD.value and not o.errors] wine_ids = list(set(o.wine_db_id for o in valid_outputs if o.wine_db_id)) if not wine_ids: return wines = self.wine_service.get_wines_with_description(db=db, ids=wine_ids) wine_map = {w.id: w.label for w in wines} for o in valid_outputs: if o.wine_db_id in wine_map: o.wine_name = wine_map[o.wine_db_id]