#! /usr/bin/env python import sys from contextlib import contextmanager import time sys.path.insert(0, '.') from datetime import datetime import time import requests from pymysql.cursors import Cursor import pymysql from tracker import soundcloud from tracker.youtube import youtube from tracker.config import _get_secret_json import traceback unicron_conf_dict = _get_secret_json('unicronSecretsProd') mysql_kwargs = dict( host=unicron_conf_dict['UC_MYSQL_HOST'], user='unicron', password=unicron_conf_dict['UC_MYSQL_PASSWORD'], db='unicron', charset='utf8mb4', # cursorclass='DictCursor' ) SC_TABLE_NAME = 'carsick_sc_plays' SC_TRACK1 = 857438392 SC_TRACK2 = None SC_TRACK3 = None YT_TABLE_NAME = 'carsick_youtubes' YT_VIDEO_1 = 'xFN0h-jc3mM' # '3xCONvov-MI' YT_VIDEO_2 = '6tHNR5DxUQI' YT_VIDEO_3 = None TIKTOK_TABLE_NAME = 'carsick_tiktoks' TIKTOK_URL1 = 'https://www.tiktok.com/music/carsick-6860336772971251714' # 'https://www.tiktok.com/music/Before-I-Go-6815672448634062849' TIKTOK_URL2 = None TIKTOK_URL3 = None class fakecur: execute = lambda *args: print("Would execute\n\n", args[0], "\n\n", args[1] if len(args) > 1 else '') @contextmanager def commit_after() -> Cursor: connection = pymysql.connect(**mysql_kwargs) try: with connection.cursor() as cur: yield cur connection.commit() finally: try: connection.close() except Exception as e: print("Warning, failed to close connection") print(e) def capture_soundcloud(): if not SC_TRACK1: print("No SoundCloud") return track_id_strs = [str(i) for i in (SC_TRACK1, SC_TRACK2, SC_TRACK3) if i is not None] try: datalist = list(sorted( soundcloud.client.get("/tracks", limit=5, ids=",".join(track_id_strs)), key=lambda r: r.id )) except Exception as e: print(f"Failed to fetch soundcloud data {e}") traceback.print_exc() return columns = ["ts"] args = [datetime.utcnow().isoformat()] for i, data in enumerate(datalist, start=1): columns.extend([f"plays_{i}", f"comments_{i}", f"views_{i}"]) args.extend([ data.playback_count, data.comment_count, data.favoritings_count, ]) pcts = ["%s"] * len(columns) sql = f'''insert into {SC_TABLE_NAME} ({",".join(columns)}) values ({",".join(pcts)})''' with commit_after() as cur: print("Recording soundcloud", args) cur.execute(sql, args) def capture_youtubes(): all_ytids = list(filter(None, [YT_VIDEO_1, YT_VIDEO_2, YT_VIDEO_3])) if not all_ytids: print("No youtubes") return try: items = list(sorted( youtube._fetch_video_stats(video_ids=all_ytids)['items'], key=lambda x: all_ytids.index(x['id']) )) except Exception as e: print(f"Failed to fetch youtube data {e}") traceback.print_exc() return columns = ["ts"] args = [datetime.utcnow().isoformat()] for i, item in enumerate(items, start=1): columns.extend([f"views{i}", f"likes{i}", f"dislikes{i}", f"comments{i}"]) args.extend([ item['statistics']['viewCount'], item['statistics']['likeCount'], item['statistics']['dislikeCount'], item['statistics']['commentCount'], ]) pcts = ["%s"] * len(columns) sql = f'''insert into {YT_TABLE_NAME} ({",".join(columns)}) values ({",".join(pcts)})''' with commit_after() as cur: print("Recording youtube", args) cur.execute(sql, args) def capture_tiktoks(): if not TIKTOK_URL1: print("No tiktoks") return video_count1 = _scrape_video_count(TIKTOK_URL1) if TIKTOK_URL2: time.sleep(5) video_count2 = _scrape_video_count(TIKTOK_URL2) else: video_count2 = 0 if TIKTOK_URL3: time.sleep(5) video_count3 = _scrape_video_count(TIKTOK_URL3) else: video_count3 = 0 sql = f''' insert into {TIKTOK_TABLE_NAME} (ts, videos1, videos2, videos3) values (%s, %s, %s, %s) ''' args = (datetime.utcnow().isoformat(), video_count1, video_count2, video_count3) with commit_after() as cur: print("Recording tiktoks", args) cur.execute(sql, args) def main_loop(scrape_period_seconds): starttime = time.time() while True: capture_soundcloud() capture_youtubes() capture_tiktoks() time.sleep(scrape_period_seconds - ((time.time() - starttime) % scrape_period_seconds)) import json from bs4 import BeautifulSoup def _scrape_video_count(url): req = requests.get(url, headers={'User-Agent': 'chrome'}) try: if url not in req.url: raise RuntimeError(f"Not on expected page, url {req.url}, {req.status_code}, {req.reason}") soup = BeautifulSoup(req.content, 'html5lib') num_toks = int(next(filter(None, [json.loads(d.text).get('props',{}).get('pageProps') for d in soup.select('script[type*="json"]')])).get('musicData', {}).get('posts')) except Exception as e: print(f"Failed to get number of tiktoks {e}") import traceback traceback.print_exc() return None else: return num_toks def initialise_tables_once(): with commit_after() as cur: print("Initialising ", SC_TABLE_NAME) cur.execute(f''' create table {SC_TABLE_NAME} ( ts timestamp not null primary key, plays_1 bigint not null, plays_2 bigint not null, comments_1 bigint not null, comments_2 bigint not null, views_1 bigint not null, views_2 bigint not null );''') print("Initialising ", YT_TABLE_NAME) cur.execute(f''' create table {YT_TABLE_NAME} ( ts timestamp not null primary key, views1 int default 0 not null, views2 int default 0 not null, views3 int default 0 not null, likes1 int default 0 not null, likes2 int default 0 not null, likes3 int default 0 not null, dislikes1 int default 0 not null, dislikes2 int default 0 not null, dislikes3 int default 0 not null, comments1 int default 0 not null, comments2 int default 0 not null, comments3 int default 0 not null );''') print("Initialising ", TIKTOK_TABLE_NAME) cur.execute(f''' create table {TIKTOK_TABLE_NAME} ( ts timestamp not null primary key, videos int not null );''') if __name__ == '__main__': scrape_period_seconds = 20.0*60.0 print(f"Launching eyes blue getter, every {scrape_period_seconds} seconds") main_loop(scrape_period_seconds)