CRM_ENVIROMENT/CRM_ENVR
7
0
Fork 0
mirror of https://git.coastit.co.za/caitlin/CRM_ENVR.git synced 2026-07-10 03:05:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

Updated

Browse Source
This commit is contained in:
Ace
2026-06-24 12:30:22 +02:00
parent b2a2f7e40f
commit 8a6ad5c6c6
1 changed files with 209 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
browser-use-service/main.py
+209 -43
View File
@@ -37,10 +37,72 @@ BROAD_KEYWORDS = [
"want someone", "need help with",
]
OFFER_PATTERNS = [
r"\bfree\s+domain\b",
r"\bweb\s+hosting\b",
r"\bfree\s+website\b",
r"\bprofessional\s+website",
r"\baffordable\s+web\b",
r"\bget\s+a\s+free\b",
r"\bsee\s+more\b",
r"\bbook\s+your\b",
r"\bhire\s+a\s+freelancer\b",
r"\bcontent\s+strategy\b",
r"\bdigital\s+marketing\b",
r"\bseo\s+service",
r"\bsocial\s+media\b",
r"\blimited\s+time\s+offer\b",
r"\bspecial\s+offer\b",
r"\bsign\s+up\s+now\b",
r"\br\d[\d,.]",
r"\bstarting\s+at\b",
r"\bper\s+month\b",
r"\bmonthly\b",
r"\bfreelancer\s+marketplace\b",
r"\bfirst\s+order\b",
r"\bdomain\s+name\b",
r"\bregister\s+your\s+domain\b",
r"\bsponsored\b",
r"\bpromoted\b",
r"\badvertisement\b",
]
REQUEST_PATTERNS = [
r"\bi\s+need\b",
r"\bi\s+need\s+someone\b",
r"\bi\s+need\s+a\b",
r"\bi\s+need\s+an\b",
r"\bi\s+want\b",
r"\bi\s+would\s+like\b",
r"\bi'm\s+looking\b",
r"\bi\s+am\s+looking\b",
r"\blooking\s+for\b",
r"\blooking\s+to\b",
r"\bwho\s+can\b",
r"\bcan\s+someone\b",
r"\bhelp\s+me\b",
r"\bneed\s+help\b",
r"\bwould\s+like\s+to\b",
r"\bwant\s+to\b",
r"\banyone\s+know\b",
r"\bdoes\s+anyone\b",
r"\brecommend\b",
r"\bsuggestion\b",
r"\bquote\s+(please|for|to)\b",
]
def kw_match(text: str) -> bool:
t = text.lower()
return any(kw in t for kw in BROAD_KEYWORDS)
def is_request(text: str) -> bool:
t = text.lower()
return any(re.search(p, t) for p in REQUEST_PATTERNS)
def is_offer(text: str) -> bool:
t = text.lower()
return any(re.search(p, t) for p in OFFER_PATTERNS)
FB_SEARCHES = [
"looking for web developer",
"need a website designed",
@@ -50,10 +112,19 @@ FB_SEARCHES = [
"looking for someone to create website",
"need ecommerce website built",
"want to hire web developer",
"website quote please",
"need wordpress website",
"I need a website for my business",
"need a site for my business",
"looking for website designer South Africa",
"need website for my small business",
"who can build me a website",
"want to create a website for my business",
"looking for affordable web design",
"need a web developer South Africa",
"help me build a website",
"need someone to design my website",
"looking for web designer near me",
"need a website for my startup",
]
VIEWPORTS = [
@@ -136,52 +207,96 @@ def _parse_fb_date(block: list[str]) -> str:
pass
return datetime.now().strftime('%Y-%m-%d')
def _extract_posts_from_text(raw: str, url: str) -> list[dict]:
lines = [l.strip() for l in raw.split('\n')]
blocks = []
cur = []
fb_run = 0
for l in lines:
if 'facebook' in l.lower():
fb_run += 1
if cur and fb_run >= 2:
blocks.append(cur)
cur = []
else:
fb_run = 0
if l and len(l) >= 15:
words = re.findall(r'[A-Za-z]{2,}', l)
if len(words) >= 2:
cur.append(l)
if cur:
blocks.append(cur)
def _clean_fb_text(text: str) -> str:
cleaned_lines = []
for l in text.split('\n'):
stripped = l.strip()
if not stripped:
continue
if len(stripped) < 3:
continue
if all(not c.isalpha() for c in stripped):
continue
cleaned_lines.append(stripped)
return '\n'.join(cleaned_lines)
def _extract_posts_from_elements(elements: list[dict], base_url: str) -> list[dict]:
posts = []
seen_texts = set()
for block in blocks:
if not block:
for el in elements:
raw_text = el.get('text', '')
if len(raw_text) < 40:
continue
kw_indices = [i for i, l in enumerate(block) if kw_match(l)]
if not kw_indices:
text = _clean_fb_text(raw_text)
if len(text) < 40:
continue
i = kw_indices[0]
start = max(0, i - 2)
end = min(len(block), i + 5)
snippet = ' '.join(block[start:end])
if len(snippet) < 40:
if is_offer(text):
continue
dekey = snippet[:80]
lines = text.split('\n')
dekey = text[:80]
if dekey in seen_texts:
continue
seen_texts.add(dekey)
request_score = 2 if is_request(text) else 0
post_url = el.get('url') or base_url
posts.append({
"title": snippet[:300],
"content": snippet[:1000],
"author": block[start] if start < i else '',
"url": url,
"title": text[:300],
"content": text[:1000],
"author": el.get('author', ''),
"url": post_url,
"source": "facebook",
"date": _parse_fb_date(block),
"date": _parse_fb_date(lines),
"_score": request_score,
})
posts.sort(key=lambda p: p.get('_score', 0), reverse=True)
for p in posts:
p.pop('_score', None)
return posts
def _extract_posts_from_text(raw: str, url: str) -> list[dict]:
lines = [l.strip() for l in raw.split('\n') if l.strip() and len(l.strip()) >= 15]
posts = []
seen_texts = set()
cur = []
for l in lines:
words = re.findall(r'[A-Za-z]{2,}', l)
if len(words) < 2:
continue
lower = l.lower()
if any(kw in lower for kw in BROAD_KEYWORDS) and not is_offer(l):
if cur:
snippet = ' '.join(cur[-3:] + [l])
if len(snippet) >= 40 and not is_offer(snippet):
dekey = snippet[:80]
if dekey not in seen_texts:
seen_texts.add(dekey)
posts.append({
"title": snippet[:300],
"content": snippet[:1000],
"author": '',
"url": url,
"source": "facebook",
"date": _parse_fb_date(cur + [l]),
})
cur = []
else:
if not is_offer(l):
snippet = l
if len(snippet) >= 40:
dekey = snippet[:80]
if dekey not in seen_texts:
seen_texts.add(dekey)
posts.append({
"title": snippet[:300],
"content": snippet[:1000],
"author": '',
"url": url,
"source": "facebook",
"date": _parse_fb_date([l]),
})
cur.append(l)
if len(cur) > 10:
cur.pop(0)
return posts
async def human_scroll(page, steps: int = None, total_delay: float = None):
@@ -212,15 +327,60 @@ async def random_idle(page):
except Exception:
pass
async def _get_article_elements(page) -> list[dict]:
return await page.evaluate('''() => {
const results = [];
const seenTexts = new Set();
const selectors = [
'div[role="article"]',
'div[role="feed"] > div',
'div.x1yztbdb',
'div[data-pagelet]',
];
for (const sel of selectors) {
const els = document.querySelectorAll(sel);
for (const el of els) {
const text = (el.innerText || '').trim();
if (text.length < 40) continue;
const key = text.substring(0, 80);
if (seenTexts.has(key)) continue;
seenTexts.add(key);
const links = el.querySelectorAll('a');
let author = '';
for (const a of links) {
const t = (a.innerText || '').trim();
if (t && t.length > 2 && t.length < 80 && /^[a-zA-ZÀ-ÿ][a-zA-ZÀ-ÿ .'-]+$/.test(t)) {
author = t;
break;
}
}
let postUrl = '';
for (const a of links) {
const h = (a.getAttribute('href') || '');
if (h.includes('/posts/') || h.includes('/photo/') || h.includes('/video/') || h.includes('/groups/')) {
postUrl = h.startsWith('http') ? h : 'https://www.facebook.com' + h;
break;
}
}
results.push({ text, author, url: postUrl });
}
if (results.length > 0) break;
}
return results;
}''')
async def search_facebook(page, query: str) -> list[dict]:
url = f'https://www.facebook.com/search/posts/?q={urllib.parse.quote(query)}'
try:
await page.goto(url, wait_until='domcontentloaded', timeout=30000)
try:
await page.wait_for_selector('div[role="article"], div[role="feed"]', timeout=15000)
except Exception:
pass
await page.wait_for_timeout(random.randint(3000, 8000))
await human_scroll(page, steps=random.randint(2, 4), total_delay=random.uniform(6, 15))
# Accidental overscroll: 10% chance to jump to bottom and back
if random.random() < 0.1:
await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
await page.wait_for_timeout(random.randint(1000, 3000))
@@ -234,11 +394,15 @@ async def search_facebook(page, query: str) -> list[dict]:
if random.random() < 0.3:
await random_idle(page)
raw = await page.evaluate('document.body.innerText')
raw_articles = await _get_article_elements(page)
posts = _extract_posts_from_elements(raw_articles, url) if raw_articles else []
if not posts:
raw = await page.evaluate('document.body.innerText')
posts = _extract_posts_from_text(raw, url)
except Exception as e:
logger.warning("Facebook search failed: %s", e)
return []
return _extract_posts_from_text(raw, url)
return posts
def cleanup_chrome():
import subprocess, signal
@@ -306,7 +470,7 @@ async def scrape_facebook(profile_path: str | None = None, force: bool = False)
return {"success": True, "leads": [], "flagged": False, "flag_reason": None, "error": None}
all_posts = []
searches = random.sample(FB_SEARCHES, k=random.randint(3, 6))
searches = random.sample(FB_SEARCHES, k=random.randint(5, 8))
for i, query in enumerate(searches):
try:
posts = await search_facebook(page, query)
@@ -417,16 +581,18 @@ Return a JSON array like ["yes","no","yes"] matching the order above."""
"need a website", "my website", "business website",
"need a designer", "help with my website",
"redesign", "update my website",
"looking for", "need a", "need an", "looking to",
"need someone", "hire a", "want someone",
"need help with", "would like", "build me",
"design my", "make me a", "create my",
]
filtered = []
for r in results:
t = r['title'].lower()
if not any(kw in t for kw in strict_keywords):
continue
if any(kw in t for kw in ['i build', 'i offer', 'i create', 'i am a', 'web developer available',
'affordable web', 'web hosting', 'free website',
'limited time', 'special offer', 'sign up now',
'offer']):
if any(kw in t for kw in ['i build website', 'i offer web', 'i am a web developer',
'affordable web design package', 'web hosting']):
continue
filtered.append(r)
return filtered[:10]