initial commit

coub.py

@@ -0,0 +1,169 @@
+from dataclasses import dataclass
+import json
+import os
+import shutil
+import subprocess
+import requests
+from bs4 import BeautifulSoup
+import typing as tp
+import time
+from tqdm import tqdm
+
+from selenium import webdriver
+from selenium.webdriver.common.keys import Keys
+from selenium.webdriver.support.ui import WebDriverWait
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.common.by import By
+
+import settings
+
+
+@dataclass
+class CoubVideo:
+    title: str
+    video_url: str
+    audio_url: str
+
+
+
+def get_login():
+    return lambda x, y: (x.send_keys(settings.LOGIN), y.send_keys(settings.PASSWORD))
+
+
+def get_sources():
+    options = webdriver.ChromeOptions()
+    options.add_argument('log-level=3')
+    options.add_argument('--ignore-ssl-errors')
+    options.add_argument("--mute-audio")
+    browser = webdriver.Chrome(options=options)
+
+    browser.get(settings.COUB_PAGE)
+    browser.find_element_by_class_name("header__login-buttons__log-in").click()
+    email = browser.find_element_by_css_selector("input[name='email'][type='email']")
+    pwd = browser.find_element_by_css_selector("input[name='password'][type='password']")
+    WebDriverWait(browser, 50).until(EC.visibility_of_element_located((By.CLASS_NAME, "registration-form__captcha")))
+    get_login()(email, pwd)
+    pwd.send_keys(Keys.RETURN)
+    WebDriverWait(browser, 200).until(EC.visibility_of_element_located((By.CLASS_NAME, "header-channel__avatar")))
+    sources = []
+    for item in settings.PARSE:
+        browser.get(settings.COUB_PAGE + item)
+        WebDriverWait(browser, 200).until(EC.visibility_of_element_located((By.CLASS_NAME, "coub__inner")))
+        SCROLL_PAUSE_TIME = 2
+
+        last_height = browser.execute_script("return document.body.scrollHeight")
+
+        while True:
+            browser.execute_script("window.scrollTo(0, document.body.scrollHeight);")
+            time.sleep(SCROLL_PAUSE_TIME)
+            sources.append(browser.page_source)
+            new_height = browser.execute_script("return document.body.scrollHeight")
+            
+            if new_height == last_height:
+                break
+            last_height = new_height
+
+        sources.append(browser.page_source) 
+    
+    browser.close()
+    return sources
+
+
+def parse_pages(pages: list) -> list:
+    links = []
+    print("Parsing source pages...")
+    for page in tqdm(pages):
+        soup = BeautifulSoup(page, 'lxml')
+        def get_links(soup) -> tp.Iterable:
+            for item in soup.find_all('div', {"class": "coub__inner"}):
+                pg = item.find('input', {"class": "viewer__copy__input"})
+                yield pg.get('value') if pg is not None else None
+            
+        links.extend([i for i in get_links(soup) if i is not None])
+    
+    return links
+        
+
+def get_video_info(link: str) -> CoubVideo:
+    data = requests.get(link.strip()).text
+
+    def get_json_from_page(page: str) -> dict:
+        lines = page.split("\n")
+        for line in lines:
+            if 'abuses' in line:
+                return json.loads(line)
+
+    info = get_json_from_page(data)
+    try:
+        title =  "".join(x for x in info.get('title') if (x.isalnum() or x in "._- "))
+        if len(title) < 1:
+            title = link.split("/view/")[-1]
+        video = info['file_versions']['html5']['video']
+        video = video.get('higher') or video.get('high')
+        audio = info['file_versions']['html5'].get('audio')
+        audio = (audio.get('high') or audio.get('med') if audio else None)
+        
+        return CoubVideo(
+            title,
+            video.get('url'),
+            audio.get('url') if audio else None
+            )
+    except:
+        print("Video can't be downloaded by this shitty script:\n\t" + link)
+    
+
+def from_json(data: list[dict]) -> list[CoubVideo]:
+    videos = []
+    for item in data:
+        videos.append(CoubVideo(
+            title=item.get('title'),
+            video_url=item.get('video_url'),
+            audio_url=item.get('audio_url')
+        ))
+    
+    return videos
+
+
+
+def to_json(videos: list[CoubVideo]) -> list[dict]:
+    result = []
+    for item in videos:
+        result.append({
+            'title': item.title,
+            'video_url': item.video_url,
+            'audio_url': item.audio_url
+        })
+    
+    return result
+
+def unique(items: list) -> list:
+    return list(dict.fromkeys(items))
+
+
+def get_videos(links: list) -> list[CoubVideo]:
+    return [get_video_info(i) for i in tqdm(links)]
+
+
+def http_get(url: str) -> requests.Response:
+    return requests.get(url, stream=True)
+
+
+def save(data: bytes, name: str, ext: str):
+    with open(f"{settings.TEMP_DIR}/{name}.{ext}", "wb") as f:
+        f.write(data)
+
+def download_video(video: CoubVideo):
+    if os.path.exists(f"{settings.OUT_DIR}/{video.title}.mp4"):
+        return
+
+    print(video.title)
+    save(http_get(video.video_url).content, video.title, "mp4")
+    save(http_get(video.audio_url).content, video.title, "mp3")
+    if video.audio_url:
+        execute_ffmpeg(video.title)
+    else:
+        shutil.copy(f"{settings.TEMP_DIR}/{video.title}.mp4", f"{settings.OUT_DIR}/{video.title}.mp4")
+
+def execute_ffmpeg(path: str):
+    proc = subprocess.Popen(["ffmpeg.exe", "-stream_loop", "-1", "-i", f"{settings.TEMP_DIR}/{path}.mp4", "-i", f"{settings.TEMP_DIR}/{path}.mp3", "-shortest", "-map", "0:v:0", "-map", "1:a:0", "-y", f"{settings.OUT_DIR}/{path}.mp4"])
+    proc.wait()