1"""
2Unified Adapters for Facebook and Amazon Scrapers
3"""
4
5import asyncio
6from typing import List, Dict
7from backend.scrapers.facebook import FacebookScraper
8from backend.scrapers.amazon import AmazonScraper
9
10async def scraper_facebook_unified(config: Dict) -> List[Dict]:
11 """Adapter for Facebook Scraper"""
12 fb_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'facebook'), {})
13
14 scraper = FacebookScraper()
15
16
17 products = await scraper.scrape(
18 query=fb_config.get('query'),
19 country=fb_config.get('country', 'ALL'),
20 max_ads=fb_config.get('max_ads', 10),
21 active_status='active'
22 )
23
24 unified_products = []
25 for p in products:
26 unified_products.append({
27 'source': 'Facebook',
28 'original_id': p['product_id'],
29 'title': p['product_name'],
30 'url': p['product_url'],
31 'price': p['price'],
32 'engagement_metric': p['evidence']['momentum']['active_ads_count'],
33 'engagement_label': 'Active Ads',
34 'opportunity_score': 0,
35
36 'metadata': {
37 'description': p['evidence']['ad_body'],
38 'image_url': p['evidence']['image_url'],
39 'cta': p['evidence']['offer']['cta_text']
40 }
41 })
42
43 print(f"✅ [Facebook] Scraped {len(unified_products)} items", flush=True)
44 return unified_products
45
46async def scraper_amazon_unified(config: Dict) -> List[Dict]:
47 """Adapter for Amazon Scraper (Enhanced with Reviews)"""
48 from backend.scrapers.amazon import AmazonScraper
49
50 amz_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'amazon'), {})
51 scraper = AmazonScraper()
52 queries = amz_config.get('keywords', ['kindle passive income'])
53 all_products = []
54
55 for query in queries:
56
57 products = await scraper.scrape(
58 query=query,
59 limit=amz_config.get('limit', 10),
60 scrape_reviews=True
61 )
62 all_products.extend(products)
63
64
65 unified_products = []
66 for p in all_products:
67 unified_products.append({
68 'source': 'Amazon',
69 'original_id': p['original_id'],
70 'title': p['title'],
71 'url': p['url'],
72 'price': p['price'],
73 'engagement_metric': p['engagement_metric'],
74 'engagement_label': 'ratings',
75 'opportunity_score': p.get('opportunity_score', 50),
76 'metadata': {
77 **p['metadata'],
78 'pain_signals': p.get('pain_signals', []),
79 'review_sentiment': p.get('review_sentiment', 'neutral'),
80 'review_count': len(p.get('reviews', []))
81 }
82 })
83
84 print(f"✅ [Amazon] Scraped {len(unified_products)} items", flush=True)
85 return unified_products
86
87async def scraper_gumroad_unified(config: Dict) -> List[Dict]:
88 """Adapter for Gumroad Scraper"""
89 from backend.scrapers.gumroad import GumroadScraper
90
91 gr_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'gumroad'), {})
92 scraper = GumroadScraper()
93
94
95 keywords = gr_config.get('keywords', ['notion template'])
96 all_products = []
97
98 for kw in keywords:
99 products = await scraper.scrape(keyword=kw, max_pages=gr_config.get('max_pages', 1))
100 all_products.extend(products)
101
102 unified_products = []
103 for p in all_products:
104 unified_products.append({
105 'source': 'Gumroad',
106 'original_id': p['product_id'],
107 'title': p['product_name'],
108 'url': p['product_url'],
109 'price': p['price'],
110 'engagement_metric': 0,
111 'engagement_label': 'sales/views',
112 'opportunity_score': p.get('opportunity_score', 50),
113 'metadata': {
114 'creator': p['creator_name'],
115 'category': p['category']
116 }
117 })
118
119 print(f"✅ [Gumroad] Scraped {len(unified_products)} items", flush=True)
120 return unified_products
121
122async def scraper_creative_market_unified(config: Dict) -> List[Dict]:
123 """Adapter for Creative Market Scraper"""
124 from backend.scrapers.creative_market import CreativeMarketScraper
125
126 cm_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'creative_market'), {})
127 scraper = CreativeMarketScraper()
128
129 keywords = cm_config.get('keywords', ['notion template'])
130 all_products = []
131
132 for kw in keywords:
133 products = await scraper.scrape(keyword=kw, max_pages=cm_config.get('max_pages', 1))
134 all_products.extend(products)
135
136 unified_products = []
137 for p in all_products:
138 unified_products.append({
139 'source': 'CreativeMarket',
140 'original_id': p['product_id'],
141 'title': p['product_name'],
142 'url': p['product_url'],
143 'price': p['price'],
144 'engagement_metric': p['sales_count'],
145 'engagement_label': 'Sales',
146 'opportunity_score': p.get('opportunity_score', 50),
147 'metadata': {
148 'creator': p['creator_name'],
149 'category': 'digital'
150 }
151 })
152 print(f"✅ [CreativeMarket] Scraped {len(unified_products)} items", flush=True)
153 return unified_products
154
155async def scraper_envato_unified(config: Dict) -> List[Dict]:
156 """Adapter for Envato Scraper"""
157 from backend.scrapers.envato import EnvatoScraper
158
159 en_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'envato'), {})
160 scraper = EnvatoScraper()
161
162 products = await scraper.scrape(
163 marketplace=en_config.get('marketplace', 'codecanyon'),
164 category=en_config.get('category', 'javascript'),
165 max_pages=en_config.get('max_pages', 1)
166 )
167
168 unified_products = []
169 for p in products:
170 unified_products.append({
171 'source': 'Envato',
172 'original_id': p['product_id'],
173 'title': p['product_name'],
174 'url': p['product_url'],
175 'price': p['price'],
176 'engagement_metric': p['sales_count'],
177 'engagement_label': 'Sales',
178 'opportunity_score': p.get('opportunity_score', 50),
179 'metadata': {
180 'creator': p['author_name'],
181 'category': 'software'
182 }
183 })
184 print(f"✅ [Envato] Scraped {len(unified_products)} items", flush=True)
185 return unified_products
186
187async def scraper_instagram_unified(config: Dict) -> List[Dict]:
188 """Adapter for Instagram Scraper"""
189 from backend.scrapers.instagram import InstagramScraper
190
191 ig_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'instagram'), {})
192 scraper = InstagramScraper()
193
194 hashtags = ig_config.get('hashtags', ['business'])
195 all_products = []
196
197 for tag in hashtags:
198 products = scraper.scrape(hashtag=tag, limit=ig_config.get('limit', 10))
199 all_products.extend(products)
200
201 unified_products = []
202 for p in all_products:
203 unified_products.append({
204 'source': 'Instagram',
205 'original_id': p['canonical_key'],
206 'title': p['product_name'],
207 'url': p['product_url'],
208 'price': 0,
209 'engagement_metric': p['evidence']['engagement_rate'],
210 'engagement_label': 'Eng. Rate',
211 'opportunity_score': p.get('opportunity_score', 50),
212 'metadata': {
213 'description': p['description'],
214 'pain_severity': p['evidence']['pain_severity'],
215 'image_url': p['product_url']
216 }
217 })
218 print(f"✅ [Instagram] Scraped {len(unified_products)} items", flush=True)
219 return unified_products
220
221async def scraper_youtube_unified(config: Dict) -> List[Dict]:
222 """Adapter for YouTube Scraper"""
223 from backend.scrapers.youtube import YouTubeScraper
224
225 yt_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'youtube'), {})
226 scraper = YouTubeScraper()
227
228 queries = yt_config.get('queries', [None])
229 all_products = []
230
231 for q in queries:
232 products = scraper.scrape(query=q, limit=yt_config.get('limit', 10))
233 all_products.extend(products)
234
235 unified_products = []
236 for p in all_products:
237 unified_products.append({
238 'source': 'YouTube',
239 'original_id': p['canonical_key'],
240 'title': p['product_name'],
241 'url': p['product_url'],
242 'price': 0,
243 'engagement_metric': p['evidence']['views'],
244 'engagement_label': 'Views',
245 'opportunity_score': p.get('opportunity_score', 50),
246 'metadata': {
247 'description': p['description'],
248 'pain_score': p['evidence']['pain_score']
249 }
250 })
251 print(f"✅ [YouTube] Scraped {len(unified_products)} items", flush=True)
252 return unified_products
253
254async def scraper_producthunt_unified(config: Dict) -> List[Dict]:
255 """Adapter for Product Hunt Scraper"""
256 from backend.scrapers.producthunt import ProductHuntScraper
257
258 ph_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'producthunt'), {})
259 scraper = ProductHuntScraper()
260
261 products = scraper.scrape(
262 order=ph_config.get('order', 'VOTES'),
263 limit=ph_config.get('limit', 10)
264 )
265
266 unified_products = []
267 for p in products:
268 unified_products.append({
269 'source': 'ProductHunt',
270 'original_id': p['canonical_key'],
271 'title': p['product_name'],
272 'url': p['product_url'],
273 'price': 0,
274 'engagement_metric': p['evidence']['upvotes'],
275 'engagement_label': 'Upvotes',
276 'opportunity_score': p.get('opportunity_score', 50),
277 'metadata': {
278 'tagline': p['description'],
279 'maker': p['evidence']['maker']
280 }
281 })
282 print(f"✅ [ProductHunt] Scraped {len(unified_products)} items", flush=True)
283 return unified_products
284
285async def scraper_google_trends_unified(config: Dict) -> List[Dict]:
286 """Adapter for Google Trends"""
287 from backend.scrapers.trends import TrendsScraper
288
289 gt_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'google_trends'), {})
290 scraper = TrendsScraper()
291
292
293
294 results = scraper.scrape(
295 keyword=gt_config.get('keywords', [None])[0],
296 limit=gt_config.get('limit', 5)
297 )
298
299 unified_products = []
300 for p in results:
301 unified_products.append({
302 'source': 'GoogleTrends',
303 'original_id': p['canonical_key'],
304 'title': p['product_name'],
305 'url': p['product_url'],
306 'price': 0,
307 'engagement_metric': p['evidence']['current_interest'],
308 'engagement_label': 'Interest',
309 'opportunity_score': int(p['evidence']['timing_score'] * 10),
310 'metadata': p['evidence']
311 })
312
313 print(f"✅ [GoogleTrends] Scraped {len(unified_products)} items", flush=True)
314 return unified_products
315
316async def scraper_notion_unified(config: Dict) -> List[Dict]:
317 """Adapter for Notion Scraper"""
318 from backend.scrapers.notion import NotionScraper
319
320
321
322 scraper = NotionScraper()
323 products = await scraper.scrape(max_pages=1)
324
325 unified_products = []
326 for p in products:
327 unified_products.append({
328 'source': 'Notion',
329 'original_id': p['canonical_key'],
330 'title': p['product_name'],
331 'url': p['product_url'],
332 'price': p['price'],
333 'engagement_metric': 0,
334 'engagement_label': 'N/A',
335 'opportunity_score': 50,
336 'metadata': {
337 'creator': p['creator_name'],
338 'price_type': p['evidence']['price_raw']
339 }
340 })
341
342 print(f"✅ [Notion] Scraped {len(unified_products)} items", flush=True)
343 return unified_products
344
345
346
347
348
349async def scraper_medium_unified(config: Dict) -> List[Dict]:
350 """Mock Adapter for Medium (Go)"""
351
352 results = [
353 {"title": "How to Build a SaaS in 2026", "url": "https://medium.com/swlh/saas-2026", "claps": 1500},
354 {"title": "The Rise of AI Agents", "url": "https://medium.com/ai/agents", "claps": 3200},
355 {"title": "Passive Income with Notion", "url": "https://medium.com/startups/notion-income", "claps": 850}
356 ]
357 query = next((s['config'] for s in config['scrapers'] if s['name'] == 'medium'), {}).get('query', 'tech')
358
359 unified = []
360 for i, r in enumerate(results):
361 unified.append({
362 'source': 'Medium',
363 'original_id': f"medium_{i}",
364 'title': f"{r['title']} ({query})",
365 'url': r['url'],
366 'price': 0,
367 'engagement_metric': r['claps'],
368 'engagement_label': 'Claps',
369 'opportunity_score': 70,
370 'metadata': {'author': 'MediumWriter'}
371 })
372 print(f"✅ [Medium] Scraped {len(unified)} items (Simulated Go Wrapper)", flush=True)
373 return unified
374
375async def scraper_canva_unified(config: Dict) -> List[Dict]:
376 """Mock Adapter for Canva (Go)"""
377 results = [
378 {"title": "Social Media Kit", "usage": 5000},
379 {"title": "Business Presentation", "usage": 12000}
380 ]
381 unified = []
382 for i, r in enumerate(results):
383 unified.append({
384 'source': 'Canva',
385 'original_id': f"canva_{i}",
386 'title': r['title'],
387 'url': "https://canva.com/templates/example",
388 'price': 0,
389 'engagement_metric': r['usage'],
390 'engagement_label': 'Uses',
391 'opportunity_score': 65,
392 'metadata': {'format': 'Template'}
393 })
394 print(f"✅ [Canva] Scraped {len(unified)} items (Simulated Go Wrapper)", flush=True)
395 return unified
396
397async def scraper_lemon_squeezy_unified(config: Dict) -> List[Dict]:
398 """Mock Adapter for Lemon Squeezy (Go)"""
399 unified = [{
400 'source': 'LemonSqueezy',
401 'original_id': 'ls_1',
402 'title': 'SaaS Starter Kit',
403 'url': 'https://lemonsqueezy.com/store/example',
404 'price': 49.00,
405 'engagement_metric': 120,
406 'engagement_label': 'Sales',
407 'opportunity_score': 80,
408 'metadata': {'revenue': '$5000+'}
409 }]
410 print(f"✅ [Lemon Squeezy] Scraped {len(unified)} items (Simulated Go Wrapper)", flush=True)
411 return unified
412
413async def scraper_wp_plugins_unified(config: Dict) -> List[Dict]:
414 """Mock Adapter for WP Plugins (Go)"""
415 unified = [{
416 'source': 'WP Plugins',
417 'original_id': 'wp_1',
418 'title': 'AI Content Generator for WP',
419 'url': 'https://wordpress.org/plugins/ai-gen',
420 'price': 0,
421 'engagement_metric': 50000,
422 'engagement_label': 'Installs',
423 'opportunity_score': 85,
424 'metadata': {'rating': '4.8/5'}
425 }]
426 print(f"✅ [WP Plugins] Scraped {len(unified)} items (Simulated Go Wrapper)", flush=True)
427 return unified
428
429async def scraper_shopify_apps_unified(config: Dict) -> List[Dict]:
430 """Adapter for Shopify App Store"""
431 from backend.scrapers.shopify_apps import ShopifyAppsScraper
432
433 sa_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'shopify_apps'), {})
434 scraper = ShopifyAppsScraper()
435
436 keyword = sa_config.get('keywords', [None])[0] if sa_config.get('keywords') else None
437 products = await scraper.scrape(
438 keyword=keyword,
439 max_apps=sa_config.get('limit', 10)
440 )
441
442 unified_products = []
443 for p in products:
444 unified_products.append({
445 'source': 'ShopifyApps',
446 'original_id': p['canonical_key'],
447 'title': p['product_name'],
448 'url': p['product_url'],
449 'price': p.get('price', 0.0),
450 'engagement_metric': p.get('install_count', 0),
451 'engagement_label': 'Installs',
452 'opportunity_score': min(70 + len(p.get('pain_signals', [])) * 5, 100),
453 'metadata': {
454 'rating': p.get('rating', 0.0),
455 'review_count': p.get('review_count', 0),
456 'developer': p.get('developer', 'Unknown'),
457 'pain_signals': p.get('pain_signals', [])
458 }
459 })
460
461 print(f"✅ [Shopify Apps] Scraped {len(unified_products)} items", flush=True)
462 return unified_products
463
464
465async def scraper_shopify_products_unified(config: Dict) -> List[Dict]:
466 """Adapter for Shopify Products Scraper (Individual Store Products)"""
467 from backend.scrapers.shopify_products import ShopifyProductsScraper
468
469 sp_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'shopify_products'), {})
470
471 scraper = ShopifyProductsScraper()
472
473
474 stores = sp_config.get('stores', None)
475 max_products = sp_config.get('limit', 20)
476
477 products = await scraper.scrape(
478 stores=stores,
479 max_products_per_store=max_products
480 )
481
482 unified_products = []
483 for p in products:
484
485 score = 50
486
487
488 if p.get('is_digital', False):
489 score += 20
490
491
492 if p.get('compare_at_price') and p.get('compare_at_price') > p.get('price', 0):
493 score += 10
494
495
496 price = p.get('price', 0)
497 if 10 <= price <= 100:
498 score += 15
499
500
501 if not p.get('available', True):
502 score -= 20
503
504 unified_products.append({
505 'source': 'ShopifyProducts',
506 'original_id': p['canonical_key'],
507 'title': p['product_name'],
508 'url': p['product_url'],
509 'price': p.get('price', 0.0),
510 'engagement_metric': p.get('inventory_quantity', 0),
511 'engagement_label': 'Inventory',
512 'opportunity_score': max(0, min(score, 100)),
513 'metadata': {
514 'store': p.get('store_domain', ''),
515 'is_digital': p.get('is_digital', False),
516 'product_type': p.get('product_type', ''),
517 'vendor': p.get('vendor', ''),
518 'tags': p.get('tags', []),
519 'available': p.get('available', True),
520 'compare_at_price': p.get('compare_at_price'),
521 'images': p.get('images', [])[:3]
522 }
523 })
524
525 print(f"✅ [Shopify Products] Scraped {len(unified_products)} items", flush=True)
526 return unified_products
527
528
529async def scraper_linkedin_unified(config: Dict) -> List[Dict]:
530 """Adapter for LinkedIn Scraper (B2B Pain Points)"""
531 from backend.scrapers.linkedin_search import LinkedInSearchScraper
532
533 li_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'linkedin'), {})
534
535 scraper = LinkedInSearchScraper()
536
537
538 query = li_config.get('query', None)
539 limit = li_config.get('limit', 10)
540
541 posts = await scraper.scrape(query=query, limit=limit)
542
543 unified_products = []
544 for p in posts:
545
546 score = 60
547
548
549 desc_lower = p.get('description', '').lower()
550 b2b_keywords = ['saas', 'b2b', 'enterprise', 'crm', 'tool', 'platform']
551 if any(kw in desc_lower for kw in b2b_keywords):
552 score += 15
553
554
555 pain_keywords = ['struggling', 'frustrated', 'need', 'looking for']
556 if any(kw in desc_lower for kw in pain_keywords):
557 score += 20
558
559 unified_products.append({
560 'source': 'LinkedIn',
561 'original_id': p['canonical_key'],
562 'title': p['product_name'],
563 'url': p['product_url'],
564 'price': 0,
565 'engagement_metric': 0,
566 'engagement_label': 'Professional Network',
567 'opportunity_score': min(score, 100),
568 'metadata': {
569 'description': p.get('description', ''),
570 'platform': 'linkedin',
571 'type': 'b2b_pain_point'
572 }
573 })
574
575 print(f"✅ [LinkedIn] Scraped {len(unified_products)} items", flush=True)
576 return unified_products
577
578
579async def scraper_reddit_unified(config: Dict) -> List[Dict]:
580 """Adapter for Reddit Scraper (Community Pain Points)"""
581 from backend.scrapers.reddit import RedditScraper
582
583 reddit_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'reddit'), {})
584
585 scraper = RedditScraper()
586
587
588 subreddits = reddit_config.get('subreddits', None)
589 limit = reddit_config.get('limit', 10)
590 time_filter = reddit_config.get('time_filter', 'week')
591
592 posts = scraper.scrape(subreddits=subreddits, limit=limit, time_filter=time_filter)
593
594 unified_products = []
595 for p in posts:
596
597 evidence = p.get('evidence', {})
598 pain_severity = evidence.get('pain_severity', 0)
599 validation_score = evidence.get('validation_score', 0)
600 monetization_potential = evidence.get('monetization_potential', 0)
601
602
603 opportunity_score = min(
604 (pain_severity * 3) + (validation_score * 2) + (monetization_potential * 5),
605 100
606 )
607
608 unified_products.append({
609 'source': 'Reddit',
610 'original_id': p['canonical_key'],
611 'title': p['product_name'],
612 'url': p['product_url'],
613 'price': 0,
614 'engagement_metric': evidence.get('upvotes', 0),
615 'engagement_label': 'Upvotes',
616 'opportunity_score': int(opportunity_score),
617 'metadata': {
618 'description': p.get('description', ''),
619 'subreddit': evidence.get('subreddit', ''),
620 'comments': evidence.get('comments', 0),
621 'pain_severity': pain_severity,
622 'validation_score': validation_score,
623 'monetization_potential': monetization_potential
624 }
625 })
626
627 print(f"✅ [Reddit] Scraped {len(unified_products)} items", flush=True)
628 return unified_products
629
630
631async def scraper_twitter_unified(config: Dict) -> List[Dict]:
632 """Adapter for Twitter/X Scraper (Real-time Pain Signals)"""
633 from backend.scrapers.twitter import TwitterScraper
634
635 twitter_config = next((s['config'] for s in config['scrapers'] if s['name'] == 'twitter'), {})
636
637 scraper = TwitterScraper()
638
639
640 query = twitter_config.get('query', None)
641 limit = twitter_config.get('limit', 10)
642
643 tweets = await scraper.scrape(query=query, limit=limit)
644
645 unified_products = []
646 for t in tweets:
647
648 evidence = t.get('evidence', {})
649 pain_count = evidence.get('pain_count', 0)
650
651 score = 50
652
653
654 score += min(pain_count * 15, 40)
655
656
657 desc_lower = t.get('description', '').lower()
658 trending_keywords = ['viral', 'trending', 'everyone', 'thousands']
659 if any(kw in desc_lower for kw in trending_keywords):
660 score += 10
661
662 unified_products.append({
663 'source': 'Twitter',
664 'original_id': t['canonical_key'],
665 'title': t['product_name'],
666 'url': t['product_url'],
667 'price': 0,
668 'engagement_metric': pain_count,
669 'engagement_label': 'Pain Signals',
670 'opportunity_score': min(score, 100),
671 'metadata': {
672 'description': t.get('description', ''),
673 'pain_signals': t.get('pain_signals', []),
674 'platform': 'twitter'
675 }
676 })
677
678 print(f"✅ [Twitter] Scraped {len(unified_products)} items", flush=True)
679 return unified_products