| |
| import httpx |
| import re |
| import asyncio |
| from datetime import datetime |
|
|
| class TiktokDl: |
| def __init__(self): |
| self.api_url = "https://tikwm.com/api/" |
| self.headers = { |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36" |
| } |
|
|
| self.meta_info = { |
| "powered_by": "devily", |
| "version": "0.6.0", |
| "support": "https://t.me/Dev_Celeste" |
| } |
|
|
| def _format_size(self, size_bytes): |
| if not size_bytes: return "0 MB" |
| return f"{round(size_bytes / (1024 * 1024), 2)} MB" |
|
|
| def _estimate_audio_size(self, duration, audio_type="mp3"): |
| if not duration: return "0 MB" |
| if audio_type == "mp3": |
| size_bytes = duration * 192 * 1024 / 8 |
| else: |
| size_bytes = duration * 44100 * 2 * 2 |
| return self._format_size(size_bytes) |
|
|
| async def fetch_video(self, tiktok_url: str, tik_direct, hd: int = 1): |
| try: |
| params = {"url": tiktok_url, "hd": hd} |
| async with httpx.AsyncClient(timeout=15.0, headers=self.headers) as client: |
| response = await client.get(self.api_url, params=params) |
| if response.status_code != 200: |
| return {"ok": False, "error": "TikWM Connection Error"} |
| |
| raw = response.json() |
| if raw.get("code") != 0: |
| return {"ok": False, "error": raw.get("msg")} |
|
|
| d = raw["data"] |
| duration = d.get("duration", 0) |
| |
| source_video_url = d.get("hdplay") or d.get("play") |
| |
| token_video = tik_direct.create_instant_token(source_video_url, "mp4") |
| token_audio_mp3 = tik_direct.create_instant_token(source_video_url, "mp3_convert") |
| token_audio_wav = tik_direct.create_instant_token(source_video_url, "wav_convert") |
| |
| token_cover = tik_direct.create_instant_token(d.get("cover"), "jpg") |
| token_dynamic_cover = tik_direct.create_instant_token(d.get("ai_dynamic_cover"), "jpg") |
| token_avatar = tik_direct.create_instant_token(d.get("author", {}).get("avatar"), "jpg") |
|
|
| return { |
| "ok": True, |
| "status": "success", |
| "processed_at": datetime.now().isoformat(), |
| "meta": self.meta_info, |
| "video_info": { |
| "id": d.get("id"), |
| "region": d.get("region"), |
| "title": d.get("title", ""), |
| "cover": token_cover, |
| "dynamic_cover": token_dynamic_cover, |
| "duration": duration, |
| "created_at": datetime.fromtimestamp(d.get("create_time")).strftime('%Y-%m-%d %H:%M:%S') if d.get("create_time") else None, |
| "is_ad": d.get("is_ad", False) |
| }, |
| "download_links": { |
| "video": token_video, |
| "audio_mp3": token_audio_mp3, |
| "audio_wav": token_audio_wav, |
| "direct_cover": token_cover, |
| "note": "CaC" |
| }, |
| "music_info": { |
| "id": d.get("music_info", {}).get("id"), |
| "title": d.get("music_info", {}).get("title"), |
| "author": d.get("music_info", {}).get("author"), |
| "duration": d.get("music_info", {}).get("duration") |
| }, |
| "stats": { |
| "likes": d.get("digg_count"), |
| "views": d.get("play_count"), |
| "shares": d.get("share_count"), |
| "comments": d.get("comment_count"), |
| "collects": d.get("collect_count") |
| }, |
| "author": { |
| "unique_id": d.get("author", {}).get("unique_id"), |
| "nickname": d.get("author", {}).get("nickname"), |
| "avatar": token_avatar, |
| "verified": d.get("author", {}).get("verified", False) |
| }, |
| "assets_info": { |
| "mp4_size": self._format_size(d.get("hd_size")), |
| "mp3_size": self._estimate_audio_size(duration, "mp3"), |
| "wav_size": self._estimate_audio_size(duration, "wav") |
| } |
| } |
| except Exception as e: |
| return {"ok": False, "error": f"Internal Error: {str(e)}"} |
