-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmodels.py
More file actions
190 lines (163 loc) · 5.58 KB
/
Copy pathmodels.py
File metadata and controls
190 lines (163 loc) · 5.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
from dataclasses import dataclass, field
from typing import Optional
import re
# Known online-booking platforms — used in scraper to detect booking presence
BOOKING_DOMAINS = [
"yclients.com",
"dikidi.net",
"n-booking.ru",
"nbooking.ru",
"altegio.ru",
"sycret.ru",
"arnica.ru",
"reservio.com",
"booksy.com",
"sonline.io",
"meest.pro",
"cliento.ru",
"appointy.com",
"simplybook.me",
"usedbutik.ru",
"moipark.ru",
"zapisy.online",
]
# Text patterns that indicate an online-booking button/link
BOOKING_TEXT_PATTERNS = [
"записаться онлайн",
"онлайн-запись",
"онлайн запись",
"запись онлайн",
"записаться",
"забронировать",
"book online",
"online booking",
]
def _is_mobile(phone: str) -> bool:
cleaned = re.sub(r"[\s\-\(\)]", "", phone)
if cleaned.startswith("8") and len(cleaned) == 11:
cleaned = "+7" + cleaned[1:]
return bool(re.match(r"^\+79\d{9}$", cleaned))
def _normalize_wa(href: str) -> Optional[str]:
"""Return canonical wa.me link or None if not a WA link."""
if not href:
return None
href = href.strip()
if "wa.me" in href or "api.whatsapp.com/send" in href or "whatsapp.com/send" in href:
# normalise api.whatsapp.com/send?phone=79... → wa.me/79...
m = re.search(r"[?&]phone=([\d+]+)", href)
if m:
digits = re.sub(r"\D", "", m.group(1))
return f"https://wa.me/{digits}"
# already wa.me/...
m = re.search(r"wa\.me/([\d+]+)", href)
if m:
digits = re.sub(r"\D", "", m.group(1))
return f"https://wa.me/{digits}"
return href
return None
def _normalize_tg(href: str) -> Optional[str]:
"""Return canonical t.me link or None if not a TG link."""
if not href:
return None
href = href.strip()
if "t.me/" in href or "telegram.me/" in href or href.startswith("tg://"):
href = href.replace("telegram.me/", "t.me/")
if not href.startswith("http") and not href.startswith("tg://"):
href = "https://" + href
return href
return None
@dataclass
class Phone:
raw: str
is_mobile: bool = field(init=False)
def __post_init__(self):
self.is_mobile = _is_mobile(self.raw)
def __str__(self):
return self.raw
@dataclass
class Review:
card_id: str
author: str
date: str
rating: Optional[float]
text: str
@dataclass
class BusinessCard:
card_id: str
name: str
address: str
city: str
category: str
phones: list[Phone]
rating: Optional[float]
review_count: Optional[int]
working_hours: str
has_website: bool
website_url: Optional[str]
card_url: str
lat: Optional[float]
lon: Optional[float]
source: str = "2gis"
# messenger contacts
whatsapp: Optional[str] = None # canonical wa.me/... link
telegram: Optional[str] = None # canonical t.me/... link
# online booking
has_booking: bool = False # True = booking button/widget found → filter OUT
booking_url: Optional[str] = None # URL of detected booking widget/button
# pain-point keyword match (keyword search, pre-AI)
pain_point: bool = False
pain_point_quote: str = ""
# Stage 2 — AI-generated fields
review_summary: str = ""
first_message: str = ""
# Pain-point AI analysis (Ollama)
negative_reviews_count: int = 0
pain_point_match: bool = False # AI verdict (vs pain_point = keyword match)
pain_point_evidence: str = "" # quote from negative review confirming match
# Rating filter bounds used when this card was collected (for auditability)
rating_min_used: Optional[float] = None
rating_max_used: Optional[float] = None
# Generated 5-message cold-contact dialog script
script_dialog: str = ""
@property
def phones_str(self) -> str:
return "; ".join(p.raw for p in self.phones)
@property
def mobile_phones_str(self) -> str:
return "; ".join(p.raw for p in self.phones if p.is_mobile)
@property
def has_messenger(self) -> bool:
return bool(self.whatsapp or self.telegram)
def to_csv_row(self) -> dict:
return {
"card_id": self.card_id,
"source": self.source,
"name": self.name,
"address": self.address,
"city": self.city,
"category": self.category,
"phones": self.phones_str,
"mobile_phones": self.mobile_phones_str,
"rating": self.rating,
"review_count": self.review_count,
"working_hours": self.working_hours,
"has_website": self.has_website,
"website_url": self.website_url or "",
"card_url": self.card_url,
"lat": self.lat,
"lon": self.lon,
"whatsapp": self.whatsapp or "",
"telegram": self.telegram or "",
"has_booking": self.has_booking,
"booking_url": self.booking_url or "",
"pain_point": self.pain_point,
"pain_point_quote": self.pain_point_quote,
"review_summary": self.review_summary,
"first_message": self.first_message,
"negative_reviews_count": self.negative_reviews_count,
"pain_point_match": self.pain_point_match,
"pain_point_evidence": self.pain_point_evidence,
"rating_min_used": self.rating_min_used,
"rating_max_used": self.rating_max_used,
"script_dialog": self.script_dialog,
}