Auto-sync permit competitors in monitoring PF

app/services/monitor.py

@@ -70,6 +70,39 @@ def _source_label(source: str) -> str:
     return {"propertyfinder": "PropertyFinder", "bayut": "Bayut"}.get(source, source)
 
 
+def _normalize_permit(value: str | None) -> str:
+    return (value or "").strip().lower()
+
+
+def _listing_key(source: Source | str, external_id: str) -> tuple[str, str]:
+    source_value = source.value if isinstance(source, Source) else str(source)
+    return source_value, str(external_id or "")
+
+
+def _format_listing_added(project: Project, listing: CompetitorListing, *, auto: bool) -> str:
+    title = listing.title or "без названия"
+    prefix = "✅ <b>Автоматически добавлен конкурент</b>" if auto else "✅ <b>Добавлен конкурент</b>"
+    return (
+        f"{prefix} — {_source_label(listing.source.value)}\n"
+        f"{title}\n"
+        f"Цена: {_fmt_price(listing.current_price, listing.currency)}\n"
+        f"Permit: <code>{listing.permit_number or project.dld_permit or '—'}</code>\n"
+        f"{listing.url}"
+    )
+
+
+def _format_listing_removed(project: Project, listing: CompetitorListing, *, auto: bool) -> str:
+    title = listing.title or "без названия"
+    prefix = "🗑️ <b>Автоматически удален конкурент</b>" if auto else "🗑️ <b>Удален конкурент</b>"
+    return (
+        f"{prefix} — {_source_label(listing.source.value)}\n"
+        f"{title}\n"
+        f"Последняя цена: {_fmt_price(listing.current_price, listing.currency)}\n"
+        f"Permit: <code>{listing.permit_number or project.dld_permit or '—'}</code>\n"
+        f"{listing.url}"
+    )
+
+
 def add_competitor_url(db: Session, project: Project, url: str) -> tuple[CompetitorListing | None, str]:
     """User-facing entrypoint: paste a URL → create CompetitorListing for the project.
     Returns (listing, error_message). error_message is empty on success.
@@ -115,6 +148,8 @@ def add_competitor_url(db: Session, project: Project, url: str) -> tuple[Competi
         title=scraped.title,
         agent_name=scraped.agent_name,
         agency_name=scraped.agency_name,
+        permit_number=scraped.permit_number,
+        auto_discovered=False,
         current_price=scraped.price,
         currency=scraped.currency or "AED",
         status=ListingStatus.ACTIVE,
@@ -153,12 +188,120 @@ def add_competitor_urls(db: Session, project: Project, urls: list[str]) -> dict:
     return {"added": added, "skipped": skipped, "errors": errors}
 
 
+def _create_listing_from_candidate(
+    db: Session,
+    project: Project,
+    candidate: ScrapedListing,
+    *,
+    permit_number: str,
+) -> CompetitorListing:
+    now = datetime.utcnow()
+    listing = CompetitorListing(
+        project_id=project.id,
+        source=Source(candidate.source),
+        external_id=candidate.external_id or candidate.url,
+        url=candidate.url,
+        title=candidate.title,
+        agent_name=candidate.agent_name,
+        agency_name=candidate.agency_name,
+        permit_number=permit_number,
+        auto_discovered=True,
+        current_price=candidate.price,
+        currency=candidate.currency or "AED",
+        status=ListingStatus.ACTIVE,
+        first_seen_at=now,
+        last_seen_at=now,
+    )
+    db.add(listing)
+    db.flush()
+    if candidate.price is not None:
+        db.add(PriceHistory(listing_id=listing.id, price=candidate.price, recorded_at=now))
+    return listing
+
+
+def _hide_tracked_suggestions(
+    db: Session,
+    project: Project,
+    suggestions: dict[str, list[ScrapedListing]],
+) -> dict[str, list[ScrapedListing]]:
+    tracked = {
+        _listing_key(l.source, l.external_id)
+        for l in db.query(CompetitorListing).filter(CompetitorListing.project_id == project.id).all()
+    }
+    return {
+        source: [item for item in items if _listing_key(item.source, item.external_id) not in tracked]
+        for source, items in suggestions.items()
+    }
+
+
+def sync_permit_competitors(
+    db: Session,
+    project: Project,
+) -> tuple[list[str], dict[str, list[ScrapedListing]], str | None]:
+    """Auto-maintain competitor listings with the same DLD permit.
+
+    Exact-permit matches are added automatically. Previously auto-discovered
+    exact-permit listings that disappear from the next permit search are
+    deleted. Manual competitors are never auto-deleted.
+    """
+    changes: list[str] = []
+    our_permit = resolve_our_permit(project)
+    if not our_permit:
+        return changes, {"propertyfinder": [], "bayut": []}, None
+
+    normalized_permit = _normalize_permit(our_permit)
+    suggestions = suggest_similar(project, our_permit=our_permit, include_tracked=True)
+    matches = [
+        item
+        for item in suggestions["propertyfinder"]
+        if _normalize_permit(item.permit_number) == normalized_permit
+    ]
+
+    matched_keys = {_listing_key(item.source, item.external_id) for item in matches}
+    existing = {
+        _listing_key(item.source, item.external_id): item
+        for item in db.query(CompetitorListing).filter(CompetitorListing.project_id == project.id).all()
+    }
+
+    for item in matches:
+        key = _listing_key(item.source, item.external_id)
+        listing = existing.get(key)
+        if listing:
+            listing.permit_number = item.permit_number or our_permit
+            if item.title:
+                listing.title = item.title
+            if item.agent_name:
+                listing.agent_name = item.agent_name
+            if item.agency_name:
+                listing.agency_name = item.agency_name
+            continue
+
+        listing = _create_listing_from_candidate(db, project, item, permit_number=item.permit_number or our_permit)
+        existing[key] = listing
+        changes.append(_format_listing_added(project, listing, auto=True))
+
+    for listing in list(existing.values()):
+        if not listing.auto_discovered:
+            continue
+        if _normalize_permit(listing.permit_number) != normalized_permit:
+            continue
+        if _listing_key(listing.source, listing.external_id) in matched_keys:
+            continue
+        changes.append(_format_listing_removed(project, listing, auto=True))
+        db.delete(listing)
+
+    return changes, _hide_tracked_suggestions(db, project, suggestions), our_permit
+
+
 def check_project(db: Session, project: Project) -> list[str]:
     """Re-scan all tracked competitor listings for one project. Returns notification texts."""
     changes: list[str] = []
     now = datetime.utcnow()
+    sync_changes, _, _ = sync_permit_competitors(db, project)
+    changes.extend(sync_changes)
 
-    for listing in list(project.listings):
+    listings = db.query(CompetitorListing).filter(CompetitorListing.project_id == project.id).all()
+    for listing in listings:
         scraper = _scraper_for(listing.source)
         scraped = scraper.fetch_listing(listing.url)
         if scraped is None:
@@ -242,6 +385,10 @@ def _notify_owner(project: Project, changes: list[str]) -> None:
         send_message(owner.tg_chat_id, c)
 
 
+def notify_project_changes(project: Project, changes: list[str]) -> None:
+    _notify_owner(project, changes)
+
+
 def _reference_url(project: Project, source: Source) -> str | None:
     """A known listing URL in the project's building for the given source.
 
@@ -268,7 +415,12 @@ def resolve_our_permit(project: Project) -> str | None:
     return PF.get_permit(ref) if ref else None
 
 
-def suggest_similar(project: Project, our_permit: str | None = None) -> dict[str, list[ScrapedListing]]:
+def suggest_similar(
+    project: Project,
+    our_permit: str | None = None,
+    *,
+    include_tracked: bool = False,
+) -> dict[str, list[ScrapedListing]]:
     """Search PF + Bayut for listings in this project's building.
 
     Candidates that share our DLD permit are the same physical listing under a
@@ -295,8 +447,11 @@ def suggest_similar(project: Project, our_permit: str | None = None) -> dict[str
         except Exception as e:
             logger.exception("Bayut suggest failed: %s", e)
 
-    # Hide candidates already tracked for this project — and our own listing.
-    excluded = {(l.source.value, l.external_id) for l in project.listings}
+    # Hide our own listing. Tracked competitors are hidden for UI suggestions,
+    # but included for automatic permit synchronization.
+    excluded: set[tuple[str, str]] = set()
+    if not include_tracked:
+        excluded.update((l.source.value, l.external_id) for l in project.listings)
     if project.our_url:
         own_src = detect_source_from_url(project.our_url)
         m = re.search(r"(\d+)\.html", project.our_url)