"""Lead writer: creates/updates dgred clients from pipeline results.""" from __future__ import annotations import hashlib import logging import re from agent_samochodowy.config import Settings from agent_samochodowy.dgred.client import DgredClient from agent_samochodowy.dgred.notifier import MatchNotifier from agent_samochodowy.dgred.reply_proposal import generate_reply_proposal from agent_samochodowy.models import MatchResult, PartQuery, Post from agent_samochodowy.status import STATUS_MATCHED, STATUS_NO_MATCH, STATUS_REVIEW, map_status logger = logging.getLogger(__name__) class LeadWriter: """Writes pipeline results to dgred as leads.""" def __init__(self, dgred: DgredClient, settings: Settings) -> None: self.dgred = dgred self.settings = settings self.notifier = MatchNotifier(settings) # Local cache: author_profile_url → client_id self._client_cache: dict[str, int | str] = {} # Ensure our statuses exist self._status_map = dgred.ensure_statuses([ STATUS_MATCHED, STATUS_REVIEW, STATUS_NO_MATCH, ]) def write_lead( self, post: Post, query: PartQuery, result: MatchResult, ) -> None: """Write a single lead to dgred.""" status_name = map_status(result, self.settings) # 1. Find or create client (dedup by author_profile_url) client_id = self._find_or_create_client(post, query) # 2. Add note with full post + match details note_content = self._build_note(post, query, result, status_name) self.dgred.add_note(client_id, note_content) # 3. Update status status_id = self._status_map.get(status_name) if status_id: self.dgred.update_client(client_id, {"clientStatus": status_id}) # 4. For Dopasowano (67): generate reply proposal if status_name == STATUS_MATCHED and result.offers: self._save_reply_proposal(client_id, post, query, result) logger.info( "Lead written: client=%s status=%s confidence=%.2f method=%s post=%s", client_id, status_name, result.confidence, result.method, post.post_id, ) @staticmethod def _shorten_external_id(url: str) -> str: """Shorten a profile URL to fit dgred's 50-char externalId limit. Extracts the numeric FB user ID when possible (e.g. 'fb:100004747195200'), otherwise falls back to a deterministic SHA-256 prefix. """ if len(url) <= 50: return url # Try to extract numeric FB profile ID m = re.search(r'profile\.php\?id=(\d+)', url) or re.search(r'facebook\.com/(\d+)', url) if m: short = f"fb:{m.group(1)}" if len(short) <= 50: return short # Fallback: deterministic hash prefix return "fbh:" + hashlib.sha256(url.encode()).hexdigest()[:45] def _find_or_create_client(self, post: Post, query: PartQuery) -> int | str: """Find existing client by profile URL or create a new one.""" external_id = self._shorten_external_id(post.author_profile_url) # Check local cache if external_id in self._client_cache: return self._client_cache[external_id] # Check dgred existing = self.dgred.find_client_by_external_id(external_id) if existing: cid = existing.get("id", existing.get("_id")) self._client_cache[external_id] = cid return cid # Create new client custom_fields = {} if query.brand: custom_fields["marka"] = query.brand if query.model: custom_fields["model"] = query.model if query.year: custom_fields["rocznik"] = query.year if query.category: custom_fields["kategoria"] = query.category if query.engine_code: custom_fields["kod_silnika"] = query.engine_code if query.oe_numbers: custom_fields["numery_oe"] = ", ".join(query.oe_numbers) custom_fields["fb_profil"] = post.author_profile_url custom_fields["url_posta"] = post.post_url custom_fields["grupa"] = post.group client_data = { "name": post.author_name, "customFields": custom_fields, "tags": [{"tag": "FB Agent", "color": 1}], } # Include externalId in creation payload to avoid separate PUT client_data["externalId"] = external_id result = self.dgred.create_client(client_data) cid = result.get("id", result.get("_id", "dry-run-id")) self._client_cache[external_id] = cid return cid def _save_reply_proposal( self, client_id: int | str, post: Post, query: PartQuery, result: MatchResult, ) -> None: """Generate and save a reply proposal for Dopasowano leads.""" proposal = generate_reply_proposal(post, query, result) self.dgred.update_client(client_id, { "customFields": { "propozycja_odpowiedzi": proposal["reply_text"], "link_oferty": proposal["offer_url"], "link_profilu": proposal["author_profile"], }, }) logger.info( "Reply proposal saved: client=%s template=%d", client_id, proposal["template_id"], ) # E-mail notification to operator self.notifier.notify_match( post, query, result, client_id, proposal["reply_text"], ) def _build_note( self, post: Post, query: PartQuery, result: MatchResult, status_name: str, ) -> str: """Build audit note content.""" lines = [ f"**Post z grupy:** {post.group}", f"**Link:** {post.post_url}", f"**Autor:** {post.author_name}", f"**Data:** {post.timestamp.isoformat()}", "", "**Treść posta:**", post.text, "", f"**Status:** {status_name}", f"**Metoda dopasowania:** {result.method}", f"**Pewność:** {result.confidence:.0%}", ] if query.brand: lines.append(f"**Marka:** {query.brand}") if query.model: lines.append(f"**Model:** {query.model}") if query.category: lines.append(f"**Kategoria:** {query.category}") if query.engine_code: lines.append(f"**Kod silnika:** {query.engine_code}") if query.oe_numbers: lines.append(f"**Numery OE:** {', '.join(query.oe_numbers)}") if result.offers: lines.append("") lines.append("**Dopasowane oferty:**") for offer in result.offers[:5]: lines.append(f"- [{offer.title}]({offer.url}) ({offer.platform}, {offer.price} {offer.currency})") lines.append("") lines.append(f"**Uzasadnienie:** {query.reason}") return "\n".join(lines)