import os import re import time from flask import Flask, request, send_file from config import EXPIRE_SECONDS, API_LIST from utils import jsonify, validate_tiktok_url, resolve_url from cache import video_path, save_meta, load_meta, cleanup_expired from extractor import extract_media_info, download_video app = Flask(__name__) @app.route("/") def index(): """Menampilkan halaman utama dengan daftar API.""" return jsonify(API_LIST), 200 @app.route("/api/tiktok") def tiktok(): """Endpoint utama TikTok: ambil info + download video ke server.""" # Bersihkan file expired setiap ada request masuk cleanup_expired() url = request.args.get("url") debug_mode = request.args.get("debug") if not url: return jsonify({ "status": "error", "message": "Parameter 'url' wajib diisi.", "example": "/api/tiktok?url=https://www.tiktok.com/@user/video/1234567890" }), 400 # 1. Resolve redirect URL pendek (seperti vt.tiktok.com) url = resolve_url(url) # 2. Normalisasi URL photo ke video agar didukung oleh yt-dlp if "/photo/" in url: url = url.replace("/photo/", "/video/") if not validate_tiktok_url(url): return jsonify({ "status": "error", "message": "URL TikTok tidak valid.", "supported_formats": [ "https://www.tiktok.com/@user/video/1234567890", "https://vm.tiktok.com/xxxxxxx", "https://vt.tiktok.com/xxxxxxx", "https://www.tiktok.com/t/xxxxxxx" ] }), 400 # 3. Ambil metadata info = extract_media_info(url, platform="tiktok") if info.get("error"): return jsonify({"status": "error", "message": info["message"]}), 500 raw = info["raw_data"] # Endpoint debug untuk melihat raw data yt-dlp langsung if debug_mode in ["1", "true"]: return jsonify(raw), 200 # 4. Download video/slideshow mp4 ke temp file try: file_id, file_path = download_video(url) except Exception as e: return jsonify({"status": "error", "message": str(e)}), 500 # Deteksi author & nickname author = raw.get("uploader_id") if not author: author = raw.get("uploader", "") if author.startswith("@"): author = author[1:] nickname = raw.get("uploader") or raw.get("creator") or author # Deteksi audio title audio_title = raw.get("track") if not audio_title: audio_title = f"original sound - {author}" if author else "original sound" # Deteksi audio url audio_url = None formats = raw.get("formats", []) for fmt in formats: if fmt.get("vcodec") == "none" and fmt.get("acodec") != "none": audio_url = fmt.get("url") break if not audio_url: for fmt in formats: if fmt.get("ext") in ["mp3", "m4a"]: audio_url = fmt.get("url") break # Deteksi slideshow / photo dari thumbnails photo = False thumbnails = raw.get("thumbnails", []) if thumbnails: photo_urls = [t.get("url") for t in thumbnails if t.get("url")] # Jika ada lebih dari 1 thumbnail, biasanya itu adalah slide gambarnya if len(photo_urls) > 1: photo = photo_urls # Konversi durasi ke format M:SS atau H:MM:SS duration_secs = raw.get("duration") duration_str = "0:00" if duration_secs is not None: try: duration_secs = int(duration_secs) minutes = duration_secs // 60 seconds = duration_secs % 60 if minutes >= 60: hours = minutes // 60 minutes = minutes % 60 duration_str = f"{hours}:{minutes:02d}:{seconds:02d}" else: duration_str = f"{minutes}:{seconds:02d}" except Exception: duration_str = str(duration_secs) # 5. Simpan metadata ke disk (filesystem — aman lintas worker) title = raw.get("title") or raw.get("description") or "video" safe_title = re.sub(r'[^\w\s-]', '', title).strip()[:60] or "video" filename = f"{safe_title}.mp4" expires_at = time.time() + EXPIRE_SECONDS save_meta(file_id, filename, expires_at) # 6. Susun response base_url = request.host_url.rstrip("/") video_url = f"{base_url}/api/file/{file_id}" result_data = { "author": author, "nickname": nickname, "avatar": raw.get("thumbnail") or raw.get("uploader_url") or "", "title": title, "views": raw.get("view_count"), "like": raw.get("like_count"), "comment": raw.get("comment_count"), "share": raw.get("share_count") or raw.get("repost_count") or 0, "download": raw.get("download_count") or 0, "collect": raw.get("collect_count") or raw.get("favorite_count") or 0, "duration": duration_str, "audio_title": audio_title, "photo": photo, "video": video_url, "videoWM": video_url, "audio": audio_url } # Hapus key yang nilainya None agar mengikuti "kalo emang ada yang engga ada ya skip" result_data = {k: v for k, v in result_data.items() if v is not None} response_body = { "creator": "@kavionn – Pratama", "status": True, "data": result_data } return jsonify(response_body), 200 @app.route("/api/file/") def serve_file(file_id): """Serve file video yang sudah didownload.""" cleanup_expired() meta = load_meta(file_id) if not meta: return jsonify({ "status": "error", "message": "File tidak ditemukan atau sudah expired (6 jam)." }), 404 if time.time() > meta["expires_at"]: return jsonify({ "status": "error", "message": "File sudah expired." }), 410 vp = video_path(file_id) if not os.path.exists(vp): return jsonify({ "status": "error", "message": "File video tidak ditemukan di server." }), 404 return send_file( vp, mimetype="video/mp4", as_attachment=False, download_name=meta["filename"] ) @app.errorhandler(404) def not_found(e): """Handler untuk route yang tidak ditemukan.""" return jsonify({ "status": "error", "message": "Endpoint tidak ditemukan.", "available_endpoints": list(API_LIST["endpoints"].keys()), "hint": "Kunjungi / untuk melihat daftar API yang tersedia." }), 404 @app.errorhandler(500) def internal_error(e): """Handler untuk internal server error.""" return jsonify({ "status": "error", "message": "Terjadi kesalahan internal pada server." }), 500 if __name__ == "__main__": app.run(host="0.0.0.0", port=7860, debug=False)