ytdlp / app.py
kavion's picture
Fix: Support TikTok photo posts by resolving redirects and normalizing /photo/ to /video/
9b5e11b unverified
Raw
History Blame Contribute Delete
6.87 kB
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/<file_id>")
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)