The music and entertainment industry generates massive amounts of data daily—streaming numbers, chart positions, social media engagement, ticket sales, and artist metrics. For record labels, talent agencies, streaming platforms, and entertainment marketers, accessing this data at scale is essential for making informed decisions about artist signings, marketing campaigns, tour planning, and content investments.
Web scraping has become an indispensable tool for entertainment industry intelligence, enabling organizations to monitor music charts across platforms, track artist performance metrics, analyze streaming trends, and gather competitive intelligence. This comprehensive guide explores how to leverage web scraping for music and entertainment industry intelligence in 2026, covering everything from chart monitoring to artist discovery and market trend analysis.
Modern entertainment intelligence requires aggregating diverse data types from numerous sources:
The challenge lies in aggregating these fragmented data sources into unified intelligence systems that can identify trends, discover emerging artists, and inform strategic decisions. Web scraping provides the foundation for building comprehensive entertainment analytics platforms.
Tracking chart performance across multiple platforms provides essential intelligence for A&R teams, marketers, and industry analysts:
# Music chart monitoring and analysis system
import asyncio
import aiohttp
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import json
@dataclass
class ChartEntry:
position: int
previous_position: Optional[int]
peak_position: int
weeks_on_chart: int
track_name: str
artist_name: str
album_name: Optional[str]
streams: Optional[int]
sales: Optional[int]
trend: str # 'up', 'down', 'same', 'new'
@dataclass
class ChartSnapshot:
chart_name: str
platform: str
region: str
date: datetime
entries: List[ChartEntry]
total_tracks: int
class MusicChartScraper:
def __init__(self, api_key: str):
self.api_key = api_key
self.chart_history = {}
self.trend_cache = {}
async def monitor_multiple_charts(self, charts_config: List[Dict]) -> Dict[str, ChartSnapshot]:
"""
Monitor multiple music charts simultaneously
"""
tasks = []
for config in charts_config:
task = self._scrape_chart(
platform=config['platform'],
chart_type=config['chart_type'],
region=config.get('region', 'global')
)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
snapshots = {}
for config, result in zip(charts_config, results):
key = f"{config['platform']}_{config['chart_type']}_{config.get('region', 'global')}"
if isinstance(result, Exception):
print(f"Error scraping {key}: {result}")
continue
snapshots[key] = result
self.chart_history[key] = result
return snapshots
async def _scrape_chart(self, platform: str, chart_type: str,
region: str) -> ChartSnapshot:
"""
Scrape a specific music chart
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
# Map platforms to their chart URLs
chart_urls = {
'billboard': f"https://www.billboard.com/charts/{chart_type}/",
'spotify': f"https://open.spotify.com/playlist/37i9dQZEVXbLRQDuF5jeBp", # Global Top 50
'apple_music': f"https://music.apple.com/us/browse/top-charts",
'shazam': f"https://www.shazam.com/charts/top-200/{region}",
'youtube_music': f"https://music.youtube.com/charts"
}
url = chart_urls.get(platform, chart_urls['billboard'])
payload = {
"url": url,
"schema": {
"chart_entries": [
{
"position": "Current chart position (numeric)",
"previous_position": "Last week's position or null if new",
"peak_position": "Highest position ever reached",
"weeks_on_chart": "Number of weeks on chart (numeric)",
"track_name": "Song title",
"artist_name": "Primary artist name",
"featured_artists": ["List of featured artists"],
"album_name": "Album name if applicable",
"streams": "Stream count if available (numeric)",
"sales": "Sales units if available (numeric)",
"trend": "Direction: up, down, same, or new",
"trend_value": "Number of positions changed"
}
],
"chart_metadata": {
"chart_name": "Name of the chart",
"chart_date": "Chart publication date",
"region": "Geographic region",
"total_entries": "Total number of entries (numeric)"
}
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=payload) as response:
data = await response.json()
chart_data = data.get('data', {})
entries = []
for entry_data in chart_data.get('chart_entries', []):
entry = ChartEntry(
position=int(entry_data.get('position', 0)),
previous_position=self._parse_optional_int(entry_data.get('previous_position')),
peak_position=int(entry_data.get('peak_position', 0)),
weeks_on_chart=int(entry_data.get('weeks_on_chart', 0)),
track_name=entry_data.get('track_name', ''),
artist_name=entry_data.get('artist_name', ''),
album_name=entry_data.get('album_name'),
streams=self._parse_optional_int(entry_data.get('streams')),
sales=self._parse_optional_int(entry_data.get('sales')),
trend=entry_data.get('trend', 'same')
)
entries.append(entry)
metadata = chart_data.get('chart_metadata', {})
return ChartSnapshot(
chart_name=metadata.get('chart_name', chart_type),
platform=platform,
region=region,
date=datetime.now(),
entries=entries,
total_tracks=len(entries)
)
def _parse_optional_int(self, value) -> Optional[int]:
"""Parse optional integer value"""
if value is None or value == '':
return None
try:
return int(value)
except (ValueError, TypeError):
return None
def analyze_chart_trends(self, chart_key: str,
days_back: int = 30) -> Dict:
"""
Analyze trends from historical chart data
"""
history = self.chart_history.get(chart_key)
if not history:
return {}
# Calculate trend metrics
rising_tracks = []
falling_tracks = []
new_entries = []
consistent_performers = []
for entry in history.entries:
if entry.trend == 'new':
new_entries.append({
'track': entry.track_name,
'artist': entry.artist_name,
'debut_position': entry.position
})
elif entry.previous_position:
position_change = entry.previous_position - entry.position
if position_change > 5:
rising_tracks.append({
'track': entry.track_name,
'artist': entry.artist_name,
'current': entry.position,
'previous': entry.previous_position,
'change': position_change
})
elif position_change < -5:
falling_tracks.append({
'track': entry.track_name,
'artist': entry.artist_name,
'current': entry.position,
'previous': entry.previous_position,
'change': position_change
})
elif entry.weeks_on_chart > 20 and abs(position_change) <= 3:
consistent_performers.append({
'track': entry.track_name,
'artist': entry.artist_name,
'weeks': entry.weeks_on_chart,
'current': entry.position
})
return {
'rising_tracks': sorted(rising_tracks, key=lambda x: x['change'], reverse=True)[:10],
'falling_tracks': sorted(falling_tracks, key=lambda x: x['change'])[:10],
'new_entries': new_entries[:10],
'consistent_performers': sorted(consistent_performers, key=lambda x: x['weeks'], reverse=True)[:10],
'analysis_date': datetime.now().isoformat()
}
async def detect_breakout_artists(self,
min_chart_positions: int = 3,
velocity_threshold: float = 0.5) -> List[Dict]:
"""
Detect artists showing breakout momentum
"""
breakout_candidates = []
# Analyze all tracked charts
for chart_key, snapshot in self.chart_history.items():
artist_performance = {}
for entry in snapshot.entries:
artist = entry.artist_name
if artist not in artist_performance:
artist_performance[artist] = {
'tracks': [],
'total_positions': 0,
'highest_position': 1000,
'avg_position': 0
}
artist_performance[artist]['tracks'].append(entry.track_name)
artist_performance[artist]['total_positions'] += entry.position
artist_performance[artist]['highest_position'] = min(
artist_performance[artist]['highest_position'],
entry.position
)
# Calculate averages and identify breakouts
for artist, data in artist_performance.items():
if len(data['tracks']) >= min_chart_positions:
data['avg_position'] = data['total_positions'] / len(data['tracks'])
# Breakout criteria: multiple charting tracks with strong positions
if data['highest_position'] <= 20 and data['avg_position'] <= 50:
breakout_candidates.append({
'artist': artist,
'chart': chart_key,
'tracks_charting': len(data['tracks']),
'highest_position': data['highest_position'],
'average_position': round(data['avg_position'], 1),
'momentum_score': self._calculate_momentum_score(data)
})
# Sort by momentum score
breakout_candidates.sort(key=lambda x: x['momentum_score'], reverse=True)
return breakout_candidates[:20]
def _calculate_momentum_score(self, artist_data: Dict) -> float:
"""Calculate momentum score based on chart performance"""
# Higher score for more tracks, better positions
track_factor = min(artist_data['tracks'].__len__(), 5) * 10
position_factor = (100 - artist_data['avg_position']) / 100 * 50
peak_factor = (100 - artist_data['highest_position']) / 100 * 40
return track_factor + position_factor + peak_factor
Comprehensive artist intelligence requires tracking metrics across multiple platforms:
# Artist performance analytics system
class ArtistAnalyticsScraper:
def __init__(self, api_key: str):
self.api_key = api_key
self.artist_profiles = {}
self.metrics_history = {}
async def build_artist_profile(self, artist_name: str) -> Dict:
"""
Build comprehensive artist profile from multiple sources
"""
profile = {
'name': artist_name,
'profile_date': datetime.now().isoformat(),
'streaming_metrics': {},
'social_metrics': {},
'video_metrics': {},
'tour_data': {},
'discography': [],
'collaborations': [],
'brand_mentions': []
}
# Gather streaming platform data
profile['streaming_metrics'] = await self._scrape_streaming_metrics(artist_name)
# Collect social media metrics
profile['social_metrics'] = await self._scrape_social_metrics(artist_name)
# Get video performance data
profile['video_metrics'] = await self._scrape_video_metrics(artist_name)
# Scrape tour and event information
profile['tour_data'] = await self._scrape_tour_data(artist_name)
# Build discography
profile['discography'] = await self._scrape_discography(artist_name)
self.artist_profiles[artist_name] = profile
return profile
async def _scrape_streaming_metrics(self, artist_name: str) -> Dict:
"""
Scrape streaming platform metrics
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
# Search for artist on Spotify
search_url = f"https://open.spotify.com/search/{artist_name.replace(' ', '%20')}"
payload = {
"url": search_url,
"schema": {
"artist_results": [
{
"name": "Artist name",
"spotify_url": "Spotify profile URL",
"monthly_listeners": "Monthly listener count (numeric)",
"followers": "Follower count (numeric)",
"popularity_score": "Spotify popularity 0-100",
"top_cities": ["Cities with most listeners"],
"verified": "Boolean indicating verified status"
}
],
"top_tracks": [
{
"track_name": "Song title",
"streams": "Stream count (numeric)",
"playlist_count": "Number of playlist appearances"
}
]
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=payload) as response:
data = await response.json()
streaming_data = data.get('data', {})
artist_result = streaming_data.get('artist_results', [{}])[0]
return {
'monthly_listeners': artist_result.get('monthly_listeners'),
'followers': artist_result.get('followers'),
'popularity_score': artist_result.get('popularity_score'),
'top_cities': artist_result.get('top_cities', []),
'verified': artist_result.get('verified', False),
'top_tracks': streaming_data.get('top_tracks', []),
'platform': 'spotify'
}
async def _scrape_social_metrics(self, artist_name: str) -> Dict:
"""
Scrape social media metrics across platforms
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
social_data = {}
# Instagram metrics
instagram_url = f"https://www.instagram.com/{artist_name.replace(' ', '').lower()}/"
instagram_payload = {
"url": instagram_url,
"schema": {
"profile": {
"followers": "Follower count (numeric)",
"following": "Following count (numeric)",
"posts": "Post count (numeric)",
"verified": "Boolean",
"bio": "Profile bio text",
"external_link": "Link in bio"
},
"recent_posts": [
{
"likes": "Like count (numeric)",
"comments": "Comment count (numeric)",
"caption": "Post caption",
"posted_date": "Post date"
}
]
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=instagram_payload) as response:
data = await response.json()
insta_data = data.get('data', {})
profile = insta_data.get('profile', {})
recent_posts = insta_data.get('recent_posts', [])
# Calculate engagement rate
avg_likes = sum(p.get('likes', 0) for p in recent_posts[:10]) / min(len(recent_posts), 10) if recent_posts else 0
followers = profile.get('followers', 1)
engagement_rate = (avg_likes / followers * 100) if followers > 0 else 0
social_data['instagram'] = {
'followers': profile.get('followers'),
'following': profile.get('following'),
'posts': profile.get('posts'),
'verified': profile.get('verified', False),
'engagement_rate': round(engagement_rate, 2),
'avg_likes': round(avg_likes, 0)
}
# TikTok metrics
tiktok_url = f"https://www.tiktok.com/@{artist_name.replace(' ', '').lower()}"
tiktok_payload = {
"url": tiktok_url,
"schema": {
"profile": {
"followers": "Follower count",
"following": "Following count",
"likes": "Total likes",
"videos": "Video count"
}
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=tiktok_payload) as response:
data = await response.json()
tiktok_data = data.get('data', {}).get('profile', {})
social_data['tiktok'] = {
'followers': tiktok_data.get('followers'),
'following': tiktok_data.get('following'),
'total_likes': tiktok_data.get('likes'),
'videos': tiktok_data.get('videos')
}
return social_data
async def _scrape_video_metrics(self, artist_name: str) -> Dict:
"""
Scrape video platform metrics (YouTube, etc.)
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
youtube_search = f"https://www.youtube.com/results?search_query={artist_name.replace(' ', '+')}+official+channel"
payload = {
"url": youtube_search,
"schema": {
"channel": {
"name": "Channel name",
"subscribers": "Subscriber count",
"total_views": "Total channel views",
"video_count": "Number of videos"
},
"recent_videos": [
{
"title": "Video title",
"views": "View count",
"uploaded": "Upload date",
"likes": "Like count"
}
]
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=payload) as response:
data = await response.json()
video_data = data.get('data', {})
return {
'youtube': {
'subscribers': video_data.get('channel', {}).get('subscribers'),
'total_views': video_data.get('channel', {}).get('total_views'),
'video_count': video_data.get('channel', {}).get('video_count'),
'recent_videos': video_data.get('recent_videos', [])
}
}
async def _scrape_tour_data(self, artist_name: str) -> Dict:
"""
Scrape concert and tour information
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
# Search on Songkick or similar
tour_url = f"https://www.songkick.com/search?query={artist_name.replace(' ', '+')}"
payload = {
"url": tour_url,
"schema": {
"upcoming_events": [
{
"date": "Event date",
"venue": "Venue name",
"city": "City",
"country": "Country",
"ticket_status": "Tickets available, sold out, etc.",
"ticket_url": "Ticket purchase link"
}
],
"past_events_count": "Number of past events",
"tour_name": "Current tour name if applicable"
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=payload) as response:
data = await response.json()
tour_data = data.get('data', {})
return {
'upcoming_events': tour_data.get('upcoming_events', []),
'past_events_count': tour_data.get('past_events_count'),
'current_tour': tour_data.get('tour_name'),
'total_upcoming': len(tour_data.get('upcoming_events', []))
}
async def _scrape_discography(self, artist_name: str) -> List[Dict]:
"""
Scrape artist discography
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
discogs_url = f"https://www.discogs.com/search/?q={artist_name.replace(' ', '+')}&type=artist"
payload = {
"url": discogs_url,
"schema": {
"releases": [
{
"title": "Release title",
"type": "Album, EP, Single, etc.",
"year": "Release year",
"label": "Record label",
"format": "CD, Vinyl, Digital, etc."
}
]
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=payload) as response:
data = await response.json()
return data.get('data', {}).get('releases', [])
Tracking live events provides insights into artist popularity and market demand:
# Concert and event intelligence system
class ConcertIntelligenceScraper:
def __init__(self, api_key: str):
self.api_key = api_key
self.event_database = {}
self.price_history = {}
async def monitor_ticket_markets(self,
artists: List[str],
venues: List[str] = None) -> Dict:
"""
Monitor ticket markets for multiple artists
"""
all_events = []
for artist in artists:
events = await self._scrape_artist_events(artist)
all_events.extend(events)
if venues:
venue_events = await self._scrape_venue_events(venues)
all_events.extend(venue_events)
# Analyze market data
market_analysis = self._analyze_ticket_market(all_events)
# Detect pricing anomalies
anomalies = self._detect_pricing_anomalies(all_events)
return {
'events': all_events,
'market_analysis': market_analysis,
'anomalies': anomalies,
'hot_shows': self._identify_hot_shows(all_events),
'monitoring_timestamp': datetime.now().isoformat()
}
async def _scrape_artist_events(self, artist: str) -> List[Dict]:
"""
Scrape events for a specific artist
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
# Scrape from multiple ticket platforms
platforms = [
f"https://www.ticketmaster.com/search?q={artist.replace(' ', '+')}",
f"https://www.stubhub.com/find/s/?q={artist.replace(' ', '+')}",
f"https://www.vividseats.com/search?searchQuery={artist.replace(' ', '+')}"
]
all_events = []
for platform_url in platforms:
payload = {
"url": platform_url,
"schema": {
"events": [
{
"event_id": "Unique event identifier",
"event_name": "Event name",
"artist": "Primary artist",
"date": "Event date",
"time": "Event time",
"venue": "Venue name",
"city": "City",
"state": "State/Province",
"country": "Country",
"min_price": "Minimum ticket price (numeric)",
"max_price": "Maximum ticket price (numeric)",
"avg_price": "Average ticket price (numeric)",
"ticket_count": "Number of tickets available",
"sold_out": "Boolean indicating sold out status",
"onsale_date": "When tickets went on sale"
}
]
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=payload) as response:
data = await response.json()
events = data.get('data', {}).get('events', [])
all_events.extend(events)
# Deduplicate events
seen = set()
unique_events = []
for event in all_events:
key = f"{event.get('artist')}_{event.get('date')}_{event.get('venue')}"
if key not in seen:
seen.add(key)
unique_events.append(event)
return unique_events
def _analyze_ticket_market(self, events: List[Dict]) -> Dict:
"""
Analyze ticket market trends
"""
if not events:
return {}
prices = [e.get('avg_price', 0) for e in events if e.get('avg_price')]
analysis = {
'total_events_tracked': len(events),
'sold_out_count': sum(1 for e in events if e.get('sold_out')),
'avg_ticket_price': round(sum(prices) / len(prices), 2) if prices else 0,
'price_range': {
'min': min(prices) if prices else 0,
'max': max(prices) if prices else 0
},
'events_by_month': {},
'top_markets': {}
}
# Group by month
for event in events:
date_str = event.get('date', '')
if date_str:
try:
month = date_str[:7] # YYYY-MM
analysis['events_by_month'][month] = analysis['events_by_month'].get(month, 0) + 1
except:
pass
# Group by market
city = event.get('city')
if city:
analysis['top_markets'][city] = analysis['top_markets'].get(city, 0) + 1
# Sort top markets
analysis['top_markets'] = dict(sorted(
analysis['top_markets'].items(),
key=lambda x: x[1],
reverse=True
)[:10])
return analysis
def _detect_pricing_anomalies(self, events: List[Dict]) -> List[Dict]:
"""
Detect unusual pricing patterns
"""
anomalies = []
for event in events:
min_price = event.get('min_price', 0)
max_price = event.get('max_price', 0)
avg_price = event.get('avg_price', 0)
# High price variance
if min_price > 0 and max_price > 0:
variance = (max_price - min_price) / min_price
if variance > 5: # 5x difference
anomalies.append({
'type': 'high_price_variance',
'event': event.get('event_name'),
'venue': event.get('venue'),
'min_price': min_price,
'max_price': max_price,
'variance_multiplier': round(variance, 1)
})
# Sudden price spike (compare with history)
event_key = f"{event.get('artist')}_{event.get('date')}"
if event_key in self.price_history:
prev_avg = self.price_history[event_key]
if prev_avg > 0:
change = (avg_price - prev_avg) / prev_avg
if change > 0.3: # 30% increase
anomalies.append({
'type': 'price_spike',
'event': event.get('event_name'),
'previous_avg': prev_avg,
'current_avg': avg_price,
'increase_percent': round(change * 100, 1)
})
self.price_history[event_key] = avg_price
return anomalies
def _identify_hot_shows(self, events: List[Dict]) -> List[Dict]:
"""
Identify high-demand shows
"""
hot_shows = []
for event in events:
score = 0
indicators = []
# Sold out
if event.get('sold_out'):
score += 50
indicators.append('sold_out')
# Low ticket count
if event.get('ticket_count', 100) < 50:
score += 30
indicators.append('low_inventory')
# High average price
if event.get('avg_price', 0) > 200:
score += 20
indicators.append('premium_pricing')
# Recent onsale (high initial demand)
onsale = event.get('onsale_date', '')
if onsale:
try:
onsale_date = datetime.fromisoformat(onsale.replace('Z', '+00:00'))
days_since_onsale = (datetime.now() - onsale_date).days
if days_since_onsale < 7 and event.get('ticket_count', 1000) < 100:
score += 25
indicators.append('fast_moving')
except:
pass
if score >= 50:
hot_shows.append({
'event': event.get('event_name'),
'artist': event.get('artist'),
'date': event.get('date'),
'venue': event.get('venue'),
'heat_score': score,
'indicators': indicators,
'avg_price': event.get('avg_price')
})
return sorted(hot_shows, key=lambda x: x['heat_score'], reverse=True)
Effective entertainment analytics requires identifying reliable data sources:
Real-time monitoring enables early detection of viral trends and emerging artists:
# Viral trend detection system
class ViralTrendDetector:
def __init__(self, api_key: str):
self.api_key = api_key
self.trend_history = {}
self.viral_thresholds = {
'tiktok': 100000, # Views per hour
'youtube': 50000,
'spotify': 10000 # Stream increase per hour
}
async def detect_viral_content(self,
monitoring_list: List[str]) -> List[Dict]:
"""
Detect potentially viral content across platforms
"""
viral_candidates = []
for track in monitoring_list:
# Check TikTok velocity
tiktok_data = await self._check_tiktok_velocity(track)
# Check YouTube velocity
youtube_data = await self._check_youtube_velocity(track)
# Check streaming velocity
streaming_data = await self._check_streaming_velocity(track)
# Calculate viral score
viral_score = self._calculate_viral_score(
tiktok_data, youtube_data, streaming_data
)
if viral_score > 70:
viral_candidates.append({
'track': track,
'viral_score': viral_score,
'platforms': {
'tiktok': tiktok_data,
'youtube': youtube_data,
'streaming': streaming_data
},
'detected_at': datetime.now().isoformat()
})
return sorted(viral_candidates, key=lambda x: x['viral_score'], reverse=True)
async def _check_tiktok_velocity(self, track: str) -> Dict:
"""
Check TikTok velocity for a track
"""
api_url = "https://papalily.p.rapidapi.com/extract"
headers = {
"X-RapidAPI-Key": self.api_key,
"X-RapidAPI-Host": "papalily.p.rapidapi.com",
"Content-Type": "application/json"
}
search_url = f"https://www.tiktok.com/search?q={track.replace(' ', '+')}"
payload = {
"url": search_url,
"schema": {
"sound_page": {
"total_videos": "Number of videos using this sound",
"total_views": "Total views across all videos"
},
"trending_videos": [
{
"views": "Video views",
"likes": "Video likes",
"shares": "Video shares",
"posted_time": "When posted"
}
]
}
}
async with aiohttp.ClientSession() as session:
async with session.post(api_url, headers=headers, json=payload) as response:
data = await response.json()
tiktok_data = data.get('data', {})
# Calculate velocity (views per hour on recent videos)
recent_videos = tiktok_data.get('trending_videos', [])[:10]
total_recent_views = sum(v.get('views', 0) for v in recent_videos)
return {
'total_videos': tiktok_data.get('sound_page', {}).get('total_videos'),
'total_views': tiktok_data.get('sound_page', {}).get('total_views'),
'recent_velocity': total_recent_views,
'is_viral': total_recent_views > self.viral_thresholds['tiktok']
}
def _calculate_viral_score(self, tiktok: Dict, youtube: Dict,
streaming: Dict) -> float:
"""
Calculate composite viral score
"""
score = 0
# TikTok weight: 40%
if tiktok.get('is_viral'):
score += 40
else:
velocity_ratio = min(tiktok.get('recent_velocity', 0) / self.viral_thresholds['tiktok'], 1)
score += velocity_ratio * 40
# YouTube weight: 30%
if youtube.get('is_viral'):
score += 30
else:
velocity_ratio = min(youtube.get('recent_velocity', 0) / self.viral_thresholds['youtube'], 1)
score += velocity_ratio * 30
# Streaming weight: 30%
if streaming.get('is_viral'):
score += 30
else:
velocity_ratio = min(streaming.get('velocity', 0) / self.viral_thresholds['spotify'], 1)
score += velocity_ratio * 30
return round(score, 1)
Entertainment data scraping requires careful attention to legal and ethical boundaries:
A complete entertainment analytics system requires robust data infrastructure:
# Complete entertainment intelligence pipeline
class EntertainmentIntelligencePipeline:
def __init__(self, api_key: str):
self.api_key = api_key
self.chart_scraper = MusicChartScraper(api_key)
self.artist_scraper = ArtistAnalyticsScraper(api_key)
self.concert_scraper = ConcertIntelligenceScraper(api_key)
self.viral_detector = ViralTrendDetector(api_key)
async def run_full_pipeline(self,
target_artists: List[str] = None,
monitor_charts: bool = True) -> Dict:
"""
Run complete entertainment intelligence pipeline
"""
results = {
'timestamp': datetime.utcnow().isoformat(),
'data_sources': {}
}
# Collect chart data
if monitor_charts:
print("Collecting music chart data...")
charts_config = [
{'platform': 'billboard', 'chart_type': 'hot-100', 'region': 'us'},
{'platform': 'billboard', 'chart_type': 'billboard-200', 'region': 'us'},
{'platform': 'spotify', 'chart_type': 'global-50', 'region': 'global'}
]
chart_data = await self.chart_scraper.monitor_multiple_charts(charts_config)
results['data_sources']['charts'] = chart_data
# Analyze trends
for chart_key in chart_data:
trends = self.chart_scraper.analyze_chart_trends(chart_key)
results['data_sources']['chart_trends'] = trends
# Collect artist data
if target_artists:
print("Collecting artist analytics...")
artist_profiles = []
for artist in target_artists:
profile = await self.artist_scraper.build_artist_profile(artist)
artist_profiles.append(profile)
results['data_sources']['artists'] = artist_profiles
# Detect breakout artists
print("Detecting breakout artists...")
breakouts = await self.chart_scraper.detect_breakout_artists()
results['insights'] = {
'breakout_artists': breakouts
}
# Generate market report
results['market_report'] = self._generate_market_report(results['data_sources'])
return results
def _generate_market_report(self, data_sources: Dict) -> Dict:
"""
Generate comprehensive market report
"""
report = {
'executive_summary': {},
'key_trends': [],
'emerging_artists': [],
'market_opportunities': []
}
# Analyze chart trends
chart_trends = data_sources.get('chart_trends', {})
if chart_trends:
report['key_trends'].append({
'category': 'chart_movement',
'rising_tracks': len(chart_trends.get('rising_tracks', [])),
'falling_tracks': len(chart_trends.get('falling_tracks', [])),
'new_entries': len(chart_trends.get('new_entries', []))
})
# Analyze artist metrics
artists = data_sources.get('artists', [])
if artists:
total_followers = sum(
a.get('social_metrics', {}).get('instagram', {}).get('followers', 0)
for a in artists
)
report['executive_summary']['total_tracked_followers'] = total_followers
return report
Stop manually collecting entertainment data from dozens of platforms. Papalily's AI-powered extraction handles music charts, artist metrics, streaming analytics, and concert data with precision. Build comprehensive entertainment intelligence in minutes, not days.
Start Entertainment Data Extraction →The music and entertainment industry thrives on data—chart positions, streaming numbers, social engagement, and live event metrics all drive critical business decisions. Web scraping provides the foundation for collecting this fragmented data at scale, enabling record labels, talent agencies, marketers, and platforms to make informed decisions about artist investments, marketing campaigns, and market opportunities.
Success in entertainment intelligence requires more than just data collection. It demands sophisticated analysis to detect emerging trends, identify viral moments before they peak, and spot breakout artists early in their trajectory. The most effective entertainment analytics platforms combine robust scraping infrastructure with intelligent trend detection and actionable insights.
As the entertainment landscape continues to evolve—with new platforms emerging and consumer behaviors shifting—the importance of comprehensive data intelligence will only grow. Organizations that invest in entertainment data scraping capabilities today will be positioned to discover the next generation of hits and artists before their competitors.
Whether you're an A&R executive searching for the next breakout star, a marketer planning a campaign, or an investor evaluating entertainment assets, automated data collection provides the competitive edge needed to succeed in the fast-moving world of music and entertainment.