sysops
110 lines
3.5 KiB
Python
110 lines
3.5 KiB
Python
"""Halal meat sales opportunity scoring for export market entities."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from typing import Any
|
||
|
||
# Land-basis: vraag naar halal vlees / moslim markt (0–100 proxy)
|
||
COUNTRY_HALAL_MARKET: dict[str, float] = {
|
||
"SA": 98, "AE": 97, "QA": 97, "KW": 96, "BH": 96, "OM": 95,
|
||
"EG": 92, "MA": 90, "DZ": 90, "TN": 89, "LY": 88, "SD": 91, "SO": 93,
|
||
"TR": 85, "ID": 94, "MY": 93, "PK": 95, "BD": 94, "AF": 96, "IR": 94,
|
||
"IQ": 93, "JO": 90, "LB": 88, "SY": 89, "YE": 95, "PS": 91,
|
||
"NG": 82, "SN": 92, "ML": 93, "NE": 95, "BF": 90, "CI": 85, "GH": 78,
|
||
"KE": 72, "TZ": 78, "UG": 70, "ET": 68, "ZA": 55, "GB": 52, "FR": 54,
|
||
"DE": 50, "NL": 58, "BE": 56, "SE": 42, "NO": 38, "DK": 40, "FI": 35,
|
||
"AT": 48, "CH": 45, "IT": 46, "ES": 48, "PT": 42, "GR": 45, "CY": 55,
|
||
"BA": 82, "AL": 85, "XK": 88, "MK": 75, "RS": 72, "BG": 58, "RO": 55,
|
||
"US": 45, "CA": 48, "AU": 38, "NZ": 35, "MX": 40, "BR": 42, "AR": 38,
|
||
"SG": 72, "TH": 55, "PH": 65, "IN": 70, "CN": 35, "JP": 30, "KR": 32,
|
||
"RU": 48, "KZ": 62, "UZ": 75, "AZ": 80, "GE": 55,
|
||
}
|
||
|
||
TYPE_BASE: dict[str, float] = {
|
||
"butcher": 92,
|
||
"doner_shoarma": 90,
|
||
"wholesaler": 86,
|
||
"distributor": 85,
|
||
"importer": 84,
|
||
"contract_caterer": 78,
|
||
"restaurant": 72,
|
||
"foodservice": 65,
|
||
"logistics": 45,
|
||
}
|
||
|
||
HALAL_KEYWORDS = re.compile(
|
||
r"halal|helal|kebab|döner|doner|shoarma|shawarma|moslim|muslim|islamic|"
|
||
r"middle\s*east|turkish|türk|arab|pakistani|moroccan|marok|syrian|lebanese|"
|
||
r"peri\s*peri|grill|bbq|meat|vlees|slager|butcher|wholesale|groothandel|"
|
||
r"cash\s*&\s*carry|horeca|ethnic|kosher",
|
||
re.I,
|
||
)
|
||
|
||
EXPLICIT_HALAL = re.compile(r"\bhalal\b|\bhelal\b", re.I)
|
||
|
||
|
||
def _product_interest_text(val: Any) -> str:
|
||
if not val:
|
||
return ""
|
||
if isinstance(val, str):
|
||
try:
|
||
val = json.loads(val)
|
||
except json.JSONDecodeError:
|
||
return val
|
||
if isinstance(val, list):
|
||
return " ".join(str(x) for x in val)
|
||
return str(val)
|
||
|
||
|
||
def score_entity(row: dict[str, Any]) -> tuple[float, list[str]]:
|
||
"""Return (score 0–100, reason tags)."""
|
||
reasons: list[str] = []
|
||
country = (row.get("country_iso2") or "").upper()
|
||
etype = row.get("entity_type") or ""
|
||
name = row.get("name") or ""
|
||
notes = row.get("halal_cert_notes") or ""
|
||
interest = _product_interest_text(row.get("product_interest"))
|
||
|
||
score = TYPE_BASE.get(etype, 50)
|
||
reasons.append(f"type:{etype}")
|
||
|
||
market = COUNTRY_HALAL_MARKET.get(country, 40)
|
||
score = score * 0.55 + market * 0.45
|
||
if market >= 80:
|
||
reasons.append("sterke_halal_markt")
|
||
|
||
blob = f"{name} {notes} {interest}".lower()
|
||
if EXPLICIT_HALAL.search(blob):
|
||
score += 18
|
||
reasons.append("halal_in_naam")
|
||
elif HALAL_KEYWORDS.search(blob):
|
||
score += 10
|
||
reasons.append("halal_signaal")
|
||
|
||
if etype in ("butcher", "doner_shoarma"):
|
||
reasons.append("direct_vlees_kanaal")
|
||
elif etype in ("wholesaler", "distributor", "importer"):
|
||
reasons.append("B2B_kanaal")
|
||
|
||
conf = int(row.get("confidence") or 50)
|
||
score += (conf - 50) * 0.08
|
||
|
||
if row.get("email") or row.get("contact_email"):
|
||
score += 3
|
||
if int(row.get("contact_count") or 0) > 0:
|
||
score += 4
|
||
reasons.append("heeft_contact")
|
||
|
||
return round(min(100, max(0, score)), 1), reasons
|
||
|
||
|
||
def halal_pin_tier(score: float) -> str:
|
||
if score >= 75:
|
||
return "hot"
|
||
if score >= 55:
|
||
return "warm"
|
||
if score >= 40:
|
||
return "mild"
|
||
return "low"
|