# app.py (Python 3.8 compatible) from flask import ( Flask, render_template, request, redirect, url_for, session, flash, jsonify, send_file, send_from_directory, abort, ) import os import secrets import shutil import uuid import time import sqlite3 from datetime import datetime from typing import Optional, Dict, Any, Tuple from config import ( FLASK_SECRET_KEY, PUBLIC_BASE_URL, MEDIA_SIGNING_SECRET, MEDIA_TOKEN_TTL_SECONDS, ADMIN_PASSWORD, ) from token_store import init_token_db, save_token from media_tokens import verify_media_token from tiktok_client import ( build_authorize_url, exchange_code_for_token, query_creator_info, upload_video_direct_post, fetch_post_status, ) from db.bulk import init_bulk_db, bulk_db from services.auth import ( require_bulk_api_key, get_access_token_for_alias, require_login_access_token, ) from services.drafts import save_draft, load_draft from services.media import ( safe_filename, video_duration_seconds_ffprobe, build_public_media_url, ) from services.creator import ( creator_can_post_now, parse_publish_at_iso_to_ts, ) from services.bulk_worker import ( resolve_access_token_for_job, process_one_scheduled_job, refresh_submitted_job_status, ) # --------------------------------------------------------------------- # Flask app # --------------------------------------------------------------------- app = Flask(__name__) app.secret_key = FLASK_SECRET_KEY app.config["PUBLIC_BASE_URL"] = PUBLIC_BASE_URL app.config["MEDIA_SIGNING_SECRET"] = MEDIA_SIGNING_SECRET app.config["MEDIA_TOKEN_TTL_SECONDS"] = MEDIA_TOKEN_TTL_SECONDS app.config["ADMIN_PASSWORD"] = ADMIN_PASSWORD BASE_DIR = os.path.dirname(os.path.abspath(__file__)) UPLOAD_DIR = os.path.join(BASE_DIR, "uploads") os.makedirs(UPLOAD_DIR, exist_ok=True) URLPROP_DIR = os.path.join(BASE_DIR, "url_properties") os.makedirs(URLPROP_DIR, exist_ok=True) BULK_DB_PATH = os.path.join(BASE_DIR, "bulk_jobs.sqlite3") init_bulk_db(BULK_DB_PATH) init_token_db() ADMIN_SESSION_KEY = "admin_authenticated" TERMINAL_JOB_STATUSES = {"complete", "failed", "cancelled"} ACTIVE_JOB_STATUSES = {"scheduled", "submitting", "submitted"} def _table_columns(conn: sqlite3.Connection, table_name: str): cur = conn.cursor() cur.execute(f"PRAGMA table_info({table_name})") cols = [] for row in cur.fetchall(): try: cols.append(row["name"]) except Exception: cols.append(row[1]) return cols def _insert_bulk_job(conn: sqlite3.Connection, job: dict) -> None: cols = _table_columns(conn, "bulk_jobs") filtered = {k: v for k, v in job.items() if k in set(cols)} colnames = ", ".join(filtered.keys()) placeholders = ", ".join(["?"] * len(filtered)) conn.execute(f"INSERT INTO bulk_jobs ({colnames}) VALUES ({placeholders})", tuple(filtered.values())) def _normalize_job_status(value: str) -> str: return (value or "").strip().lower() def _serialize_job_match(row): if not row: return None return { "job_id": row["id"], "status": row["status"], "publish_at_ts": row["publish_at_ts"], "updated_at_ts": row["updated_at_ts"], "publish_id": row["publish_id"], "last_status": row["last_status"], "last_fail_reason": row["last_fail_reason"], } def _lookup_existing_jobs_by_original_filename(*, account_alias: str, original_filename: str): """ Devuelve coincidencias por alias + nombre de archivo original. Solo los jobs activos bloquean una nueva programación. Los jobs terminales deben permitir re-subir/reprogramar el video. """ conn = bulk_db(BULK_DB_PATH) cur = conn.cursor() cur.execute( """ SELECT id, status, publish_at_ts, updated_at_ts, publish_id, last_status, last_fail_reason FROM bulk_jobs WHERE account_alias=? AND original_filename=? ORDER BY CASE WHEN LOWER(COALESCE(status, '')) IN ('scheduled', 'submitting', 'submitted') THEN 0 ELSE 1 END, publish_at_ts DESC, updated_at_ts DESC """, (account_alias, original_filename), ) rows = cur.fetchall() conn.close() active_match = None latest_terminal_match = None for row in rows: status = _normalize_job_status(row["status"]) if status in ACTIVE_JOB_STATUSES and active_match is None: active_match = row elif status in TERMINAL_JOB_STATUSES and latest_terminal_match is None: latest_terminal_match = row return { "account_alias": account_alias, "original_filename": original_filename, "matches_total": len(rows), "can_schedule": active_match is None, "active_match": _serialize_job_match(active_match), "latest_terminal_match": _serialize_job_match(latest_terminal_match), } def _is_admin_authenticated() -> bool: return session.get(ADMIN_SESSION_KEY) is True def _admin_guard(): if _is_admin_authenticated(): return None next_url = request.full_path if request.query_string else request.path return redirect(url_for("admin_login", next=next_url)) def _job_file_path(stored_filename: str): filename = (stored_filename or "").strip() if not filename or "/" in filename or "\\" in filename or ".." in filename: return None return os.path.join(UPLOAD_DIR, filename) def _job_has_file(stored_filename: str) -> bool: path = _job_file_path(stored_filename) return bool(path and os.path.isfile(path)) def _format_ts(ts) -> str: if ts in (None, ""): return "-" try: return datetime.fromtimestamp(int(ts)).strftime("%Y-%m-%d %H:%M:%S") except Exception: return str(ts) def _format_bytes(size) -> str: if size in (None, ""): return "-" units = ["B", "KB", "MB", "GB"] value = float(size) unit = units[0] for unit in units: if value < 1024 or unit == units[-1]: break value /= 1024.0 return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} {unit}" def _status_tone(status: str) -> str: st = (status or "").strip().lower() if st == "complete": return "ok" if st == "failed": return "err" if st == "cancelled": return "warn" if st in {"submitted", "submitting"}: return "warn" return "info" def _job_for_admin(row: sqlite3.Row): job = dict(row) status = (job.get("status") or "").strip().lower() job["is_terminal"] = status in TERMINAL_JOB_STATUSES job["status_tone"] = _status_tone(status) job["has_file"] = _job_has_file(job.get("stored_filename") or "") job["created_at_label"] = _format_ts(job.get("created_at_ts")) job["publish_at_label"] = _format_ts(job.get("publish_at_ts")) job["next_attempt_label"] = _format_ts(job.get("next_attempt_ts")) job["updated_at_label"] = _format_ts(job.get("updated_at_ts")) if job["has_file"]: file_path = _job_file_path(job.get("stored_filename") or "") try: job["file_size_bytes"] = os.path.getsize(file_path) except OSError: job["file_size_bytes"] = None else: job["file_size_bytes"] = None job["file_size_label"] = _format_bytes(job["file_size_bytes"]) return job def _load_admin_jobs(): conn = bulk_db(BULK_DB_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute( """ SELECT * FROM bulk_jobs ORDER BY CASE WHEN status IN ('scheduled', 'submitting', 'submitted') THEN 0 ELSE 1 END, publish_at_ts ASC, updated_at_ts DESC """ ) rows = cur.fetchall() conn.close() jobs = [] for row in rows: job = _job_for_admin(row) if job["is_terminal"] and not job["has_file"]: continue jobs.append(job) return jobs def _load_upload_inventory(jobs): job_links = {} for job in jobs: stored_filename = (job.get("stored_filename") or "").strip() if not stored_filename: continue job_links.setdefault(stored_filename, []).append(job) files = [] for entry in os.scandir(UPLOAD_DIR): if not entry.is_file(): continue stat = entry.stat() linked_jobs = job_links.get(entry.name, []) files.append( { "stored_filename": entry.name, "size_bytes": stat.st_size, "size_label": _format_bytes(stat.st_size), "modified_at_label": _format_ts(int(stat.st_mtime)), "linked_jobs_count": len(linked_jobs), "active_jobs_count": sum(0 if j["is_terminal"] else 1 for j in linked_jobs), "terminal_jobs_count": sum(1 if j["is_terminal"] else 0 for j in linked_jobs), } ) files.sort( key=lambda item: ( item["active_jobs_count"] == 0, -item["linked_jobs_count"], item["stored_filename"].lower(), ) ) return files def _count_jobs_for_file(stored_filename: str): conn = bulk_db(BULK_DB_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute( """ SELECT status, COUNT(*) AS total FROM bulk_jobs WHERE stored_filename=? GROUP BY status """, (stored_filename,), ) rows = cur.fetchall() conn.close() active = 0 terminal = 0 for row in rows: total = int(row["total"] or 0) if (row["status"] or "").strip().lower() in TERMINAL_JOB_STATUSES: terminal += total else: active += total return active, terminal # --------------------------------------------------------------------- # Media route (TikTok pull + URL properties .txt) # --------------------------------------------------------------------- @app.get("/media/") def media(token_or_file: str): # 1) Allow .txt verification files under /media/ if token_or_file.lower().endswith(".txt"): full = os.path.join(URLPROP_DIR, token_or_file) if not os.path.exists(full): abort(404) return send_from_directory(URLPROP_DIR, token_or_file, as_attachment=False, mimetype="text/plain") # 2) Normal signed token flow for mp4 payload = verify_media_token(token_or_file, app.config["MEDIA_SIGNING_SECRET"]) if not payload: abort(403) filename = payload.get("f") if not filename: abort(400) # prevent path traversal if "/" in filename or "\\" in filename or ".." in filename: abort(400) full_path = os.path.join(UPLOAD_DIR, filename) if not os.path.exists(full_path): abort(404) return send_from_directory(UPLOAD_DIR, filename, as_attachment=False, mimetype="video/mp4") # --------------------------------------------------------------------- # Pages # --------------------------------------------------------------------- @app.route("/") def landing(): is_connected = session.get("tiktok_token") is not None return render_template("landing.html", is_connected=is_connected) @app.route("/app") def app_home(): is_connected = session.get("tiktok_token") is not None return render_template("index.html", is_connected=is_connected) @app.route("/contact") def contact(): is_connected = session.get("tiktok_token") is not None return render_template("contact.html", is_connected=is_connected) @app.route("/tos") def tos(): return render_template("tos.html") @app.route("/privacy") def privacy(): return render_template("privacy.html") # --------------------------------------------------------------------- # Admin panel # --------------------------------------------------------------------- @app.route("/admin/login", methods=["GET", "POST"]) def admin_login(): if _is_admin_authenticated(): return redirect(url_for("admin_jobs")) next_url = (request.args.get("next") or request.form.get("next") or "").strip() if request.method == "POST": provided = request.form.get("password") or "" expected = str(app.config.get("ADMIN_PASSWORD") or "") if expected and secrets.compare_digest(provided, expected): session[ADMIN_SESSION_KEY] = True session.modified = True flash("Acceso admin concedido.", "success") if next_url.startswith("/"): return redirect(next_url) return redirect(url_for("admin_jobs")) flash("Contraseña admin incorrecta.", "error") return render_template("admin_login.html", next_url=next_url) @app.route("/admin/logout", methods=["POST"]) def admin_logout(): session.pop(ADMIN_SESSION_KEY, None) session.modified = True flash("Sesión admin cerrada.", "success") return redirect(url_for("admin_login")) @app.get("/admin") def admin_jobs(): guard = _admin_guard() if guard: return guard jobs = _load_admin_jobs() in_progress_jobs = [job for job in jobs if not job["is_terminal"]] finished_jobs = [job for job in jobs if job["is_terminal"]] upload_files = _load_upload_inventory(jobs) return render_template( "admin_jobs.html", in_progress_jobs=in_progress_jobs, finished_jobs=finished_jobs, upload_files=upload_files, total_jobs=len(jobs), ) @app.post("/admin/jobs//cancel") def admin_cancel_job(job_id: str): guard = _admin_guard() if guard: return guard conn = bulk_db(BULK_DB_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT * FROM bulk_jobs WHERE id=?", (job_id,)) row = cur.fetchone() if not row: conn.close() flash("El trabajo indicado no existe.", "error") return redirect(url_for("admin_jobs")) status = (row["status"] or "").strip().lower() if status in TERMINAL_JOB_STATUSES: conn.close() flash("Ese trabajo ya estaba finalizado.", "warning") return redirect(url_for("admin_jobs")) now_ts = int(time.time()) cur.execute( """ UPDATE bulk_jobs SET status='cancelled', next_attempt_ts=?, last_status='CANCELLED_BY_ADMIN', updated_at_ts=? WHERE id=? AND status NOT IN ('complete', 'failed', 'cancelled') """, (now_ts, now_ts, job_id), ) conn.commit() conn.close() if cur.rowcount != 1: flash("No se pudo cancelar el trabajo porque su estado cambió mientras tanto.", "warning") return redirect(url_for("admin_jobs")) if status in {"submitted", "submitting"}: flash( "Trabajo cancelado localmente. Si TikTok ya había aceptado el envío, esta acción no puede retirarlo allí.", "warning", ) else: flash("Trabajo cancelado correctamente.", "success") return redirect(url_for("admin_jobs")) @app.post("/admin/jobs//relaunch") def admin_relaunch_job(job_id: str): guard = _admin_guard() if guard: return guard conn = bulk_db(BULK_DB_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT * FROM bulk_jobs WHERE id=?", (job_id,)) row = cur.fetchone() if not row: conn.close() flash("El trabajo indicado no existe.", "error") return redirect(url_for("admin_jobs")) status = (row["status"] or "").strip().lower() if status not in TERMINAL_JOB_STATUSES: conn.close() flash("Solo se pueden relanzar trabajos finalizados.", "error") return redirect(url_for("admin_jobs")) source_filename = row["stored_filename"] or "" source_path = _job_file_path(source_filename) if not source_path or not os.path.isfile(source_path): conn.close() flash("No se puede relanzar porque el archivo original ya no existe.", "error") return redirect(url_for("admin_jobs")) original_filename = safe_filename((row["original_filename"] or "").strip() or source_filename) new_job_id = uuid.uuid4().hex new_stored_filename = f"{new_job_id}_{original_filename}" new_file_path = _job_file_path(new_stored_filename) try: shutil.copy2(source_path, new_file_path) except OSError as exc: conn.close() flash(f"No se pudo duplicar el archivo para el relanzamiento: {exc}", "error") return redirect(url_for("admin_jobs")) now_ts = int(time.time()) job = { "id": new_job_id, "created_at_ts": now_ts, "publish_at_ts": now_ts, "next_attempt_ts": now_ts, "status": "scheduled", "attempts": 0, "last_error": None, "access_token": row["access_token"], "account_alias": row["account_alias"], "title": row["title"], "privacy_level": row["privacy_level"], "allow_comment": row["allow_comment"], "allow_duet": row["allow_duet"], "allow_stitch": row["allow_stitch"], "commercial_toggle": row["commercial_toggle"], "brand_organic_toggle": row["brand_organic_toggle"], "brand_content_toggle": row["brand_content_toggle"], "is_aigc": row["is_aigc"], "stored_filename": new_stored_filename, "original_filename": original_filename, "video_url": "", "publish_id": None, "last_status": None, "last_fail_reason": None, "updated_at_ts": now_ts, } try: _insert_bulk_job(conn, job) conn.commit() except Exception as exc: conn.close() try: os.remove(new_file_path) except OSError: pass flash(f"No se pudo crear el nuevo trabajo: {exc}", "error") return redirect(url_for("admin_jobs")) conn.close() flash(f"Trabajo relanzado. Nuevo job: {new_job_id}", "success") return redirect(url_for("admin_jobs")) @app.post("/admin/files/delete") def admin_delete_file(): guard = _admin_guard() if guard: return guard stored_filename = (request.form.get("stored_filename") or "").strip() file_path = _job_file_path(stored_filename) if not file_path: flash("Nombre de archivo inválido.", "error") return redirect(url_for("admin_jobs")) active_jobs, terminal_jobs = _count_jobs_for_file(stored_filename) if not os.path.isfile(file_path): flash("El archivo ya no existe en el sistema.", "warning") return redirect(url_for("admin_jobs")) try: os.remove(file_path) except OSError as exc: flash(f"No se pudo borrar el archivo: {exc}", "error") return redirect(url_for("admin_jobs")) if active_jobs > 0: flash( f"Archivo borrado. Ojo: había {active_jobs} trabajo(s) en curso usando ese vídeo y ahora quedarán sin fichero.", "warning", ) elif terminal_jobs > 0: flash("Archivo borrado. Los trabajos finalizados asociados dejarán de mostrarse en el panel.", "success") else: flash("Archivo borrado del sistema.", "success") return redirect(url_for("admin_jobs")) # --------------------------------------------------------------------- # OAuth flow (UI) with alias persistence for cron # --------------------------------------------------------------------- @app.route("/login") def login(): """ Optional: /login?alias=en or /login?alias=es This alias will be used to persist tokens server-side for cron/worker. """ alias = (request.args.get("alias") or "default").strip() state = secrets.token_hex(16) session["oauth_state"] = state session["oauth_alias"] = alias return redirect(build_authorize_url(state)) @app.route("/tiktok/callback") def tiktok_callback(): error = request.args.get("error") if error: flash(f"TikTok error: {error}", "error") return redirect("/app") code = request.args.get("code") state = request.args.get("state") if not code or not state or state != session.get("oauth_state"): flash("Invalid OAuth state or missing code.", "error") return redirect("/app") try: token_data = exchange_code_for_token(code) session["tiktok_token"] = token_data session.modified = True alias = (session.get("oauth_alias") or "default").strip() save_token(alias, token_data) flash(f"Successfully connected with TikTok ({alias}).", "success") except Exception as e: flash(f"Error exchanging code: {e}", "error") return redirect("/app") @app.route("/logout", methods=["POST"]) def logout(): session.pop("tiktok_token", None) session.pop("drafts", None) flash("Disconnected.", "success") return redirect("/app") # --------------------------------------------------------------------- # Bulk API: publish now (server-to-server) # Supports access_token or account_alias (recommended) # --------------------------------------------------------------------- @app.route("/api/bulk/publish", methods=["POST"]) def api_bulk_publish(): require_bulk_api_key() account_alias = (request.form.get("account_alias") or "").strip() access_token = (request.form.get("access_token") or "").strip() if account_alias: try: access_token = get_access_token_for_alias(account_alias) except Exception as e: return jsonify({"ok": False, "error": "alias_token_failed", "details": str(e)}), 400 if not access_token: return jsonify({"ok": False, "error": "missing_access_token_or_account_alias"}), 400 if "video" not in request.files: return jsonify({"ok": False, "error": "missing_video"}), 400 file = request.files["video"] if not file.filename: return jsonify({"ok": False, "error": "missing_filename"}), 400 # Params title = (request.form.get("title") or "").strip() privacy_level = (request.form.get("privacy_level") or "").strip() if not privacy_level: return jsonify({"ok": False, "error": "missing_privacy_level"}), 400 allow_comment = request.form.get("allow_comment", "1") == "1" allow_duet = request.form.get("allow_duet", "1") == "1" allow_stitch = request.form.get("allow_stitch", "1") == "1" commercial_toggle = request.form.get("commercial_toggle", "0") == "1" brand_organic_toggle = request.form.get("brand_organic_toggle", "0") == "1" brand_content_toggle = request.form.get("brand_content_toggle", "0") == "1" is_aigc = request.form.get("is_aigc", "1") == "1" # Branded content cannot be SELF_ONLY if brand_content_toggle and privacy_level == "SELF_ONLY": return jsonify({"ok": False, "error": "branded_cannot_be_self_only"}), 400 # creator_info (latest) try: creator_info = query_creator_info(access_token) except Exception as e: return jsonify({"ok": False, "error": "creator_info_failed", "details": str(e)}), 400 can_post_now, reason = creator_can_post_now(creator_info) if not can_post_now: return jsonify({"ok": False, "error": "cannot_post_now", "reason": reason}), 400 # Validate privacy option is allowed options = creator_info.get("privacy_level_options") or [] if privacy_level not in options: return jsonify({"ok": False, "error": "invalid_privacy_level", "options": options}), 400 # Respect creator settings if creator_info.get("comment_disabled") is True: allow_comment = False if creator_info.get("duet_disabled") is True: allow_duet = False if creator_info.get("stitch_disabled") is True: allow_stitch = False disable_comment = not allow_comment disable_duet = not allow_duet disable_stitch = not allow_stitch # Save file to uploads/ original_fn = safe_filename(file.filename) draft_id = uuid.uuid4().hex stored_filename = f"{draft_id}_{original_fn}" save_path = os.path.join(UPLOAD_DIR, stored_filename) file.save(save_path) # Duration enforcement (best-effort) max_dur = creator_info.get("max_video_post_duration_sec") dur = video_duration_seconds_ffprobe(save_path) if isinstance(max_dur, int) and max_dur > 0 and dur > 0 and dur > max_dur: try: os.remove(save_path) except Exception: pass return jsonify({"ok": False, "error": "duration_exceeds_max", "dur": dur, "max": max_dur}), 400 # Build pull URL (signed) video_url = build_public_media_url( stored_filename=stored_filename, public_base_url=app.config["PUBLIC_BASE_URL"], signing_secret=app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=app.config["MEDIA_TOKEN_TTL_SECONDS"], ) # Direct Post try: init_resp = upload_video_direct_post( access_token=access_token, caption=title, privacy_level=privacy_level, disable_comment=disable_comment, disable_duet=disable_duet, disable_stitch=disable_stitch, brand_content_toggle=brand_content_toggle, brand_organic_toggle=brand_organic_toggle, is_aigc=is_aigc, mode="PULL_FROM_URL", video_url=video_url, ) publish_id = (init_resp.get("data") or {}).get("publish_id") return jsonify( { "ok": True, "publish_id": publish_id, "video_url": video_url, "init_resp": init_resp, "stored_filename": stored_filename, } ), 200 except Exception as e: return jsonify({"ok": False, "error": "direct_post_failed", "details": str(e)}), 400 # --------------------------------------------------------------------- # Bulk API: status (server-to-server) # Supports access_token or account_alias (recommended) # --------------------------------------------------------------------- @app.route("/api/bulk/status", methods=["POST"]) def api_bulk_status(): require_bulk_api_key() account_alias = (request.form.get("account_alias") or "").strip() access_token = (request.form.get("access_token") or "").strip() if account_alias: try: access_token = get_access_token_for_alias(account_alias) except Exception as e: return jsonify({"ok": False, "error": "alias_token_failed", "details": str(e)}), 400 publish_id = (request.form.get("publish_id") or "").strip() if not access_token or not publish_id: return jsonify({"ok": False, "error": "missing_access_token_or_publish_id"}), 400 try: data = fetch_post_status(access_token, publish_id) return jsonify({"ok": True, "data": data}), 200 except Exception as e: return jsonify({"ok": False, "error": str(e)}), 200 @app.route("/api/bulk/find_existing", methods=["POST"]) def api_bulk_find_existing(): require_bulk_api_key() account_alias = (request.form.get("account_alias") or "").strip() original_filename_raw = (request.form.get("original_filename") or request.form.get("filename") or "").strip() if not account_alias: return jsonify({"ok": False, "error": "account_alias_required"}), 400 if not original_filename_raw: return jsonify({"ok": False, "error": "missing_original_filename"}), 400 original_filename = safe_filename(original_filename_raw) lookup = _lookup_existing_jobs_by_original_filename( account_alias=account_alias, original_filename=original_filename, ) return jsonify({"ok": True, **lookup}), 200 # --------------------------------------------------------------------- # Bulk API: schedule job (server-to-server) # --------------------------------------------------------------------- @app.route("/api/bulk/schedule", methods=["POST"]) def api_bulk_schedule(): require_bulk_api_key() account_alias = (request.form.get("account_alias") or "").strip() access_token = (request.form.get("access_token") or "").strip() # If alias provided, resolve an access token now (to satisfy DB not-null), # but cron will refresh again at publish time. if account_alias and not access_token: try: access_token = get_access_token_for_alias(account_alias) except Exception as e: return jsonify({"ok": False, "error": "alias_token_failed", "details": str(e)}), 400 if not access_token: return jsonify({"ok": False, "error": "missing_access_token_or_account_alias"}), 400 publish_at = (request.form.get("publish_at") or "").strip() if not publish_at: return jsonify({"ok": False, "error": "missing_publish_at"}), 400 try: publish_at_ts = parse_publish_at_iso_to_ts(publish_at) except Exception as e: return jsonify({"ok": False, "error": "invalid_publish_at", "details": str(e)}), 400 if "video" not in request.files: return jsonify({"ok": False, "error": "missing_video"}), 400 file = request.files["video"] if not file.filename: return jsonify({"ok": False, "error": "missing_filename"}), 400 title = (request.form.get("title") or "").strip() privacy_level = (request.form.get("privacy_level") or "").strip() if not privacy_level: return jsonify({"ok": False, "error": "missing_privacy_level"}), 400 dedupe_mode = (request.form.get("dedupe_mode") or "active_only").strip().lower() if dedupe_mode not in {"none", "active_only", "all"}: return jsonify({"ok": False, "error": "invalid_dedupe_mode"}), 400 allow_comment = request.form.get("allow_comment", "1") == "1" allow_duet = request.form.get("allow_duet", "1") == "1" allow_stitch = request.form.get("allow_stitch", "1") == "1" commercial_toggle = request.form.get("commercial_toggle", "0") == "1" brand_organic_toggle = request.form.get("brand_organic_toggle", "0") == "1" brand_content_toggle = request.form.get("brand_content_toggle", "0") == "1" is_aigc = request.form.get("is_aigc", "1") == "1" if brand_content_toggle and privacy_level == "SELF_ONLY": return jsonify({"ok": False, "error": "branded_cannot_be_self_only"}), 400 original_fn = safe_filename(file.filename) existing_lookup = _lookup_existing_jobs_by_original_filename( account_alias=account_alias, original_filename=original_fn, ) existing_match = None if dedupe_mode == "active_only": existing_match = existing_lookup["active_match"] elif dedupe_mode == "all": existing_match = existing_lookup["active_match"] or existing_lookup["latest_terminal_match"] if existing_match: existing_status = _normalize_job_status(existing_match["status"]) return jsonify( { "ok": True, "skipped": True, "reason": "already_scheduled" if existing_status in ACTIVE_JOB_STATUSES else "already_exists", "job_id": existing_match["job_id"], "status": existing_match["status"], "publish_at_ts": existing_match["publish_at_ts"], "original_filename": original_fn, "account_alias": account_alias or None, "dedupe_mode": dedupe_mode, } ), 200 job_id = uuid.uuid4().hex stored_filename = f"{job_id}_{original_fn}" save_path = os.path.join(UPLOAD_DIR, stored_filename) file.save(save_path) video_url = build_public_media_url( stored_filename=stored_filename, public_base_url=app.config["PUBLIC_BASE_URL"], signing_secret=app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=app.config["MEDIA_TOKEN_TTL_SECONDS"], ) now_ts = int(time.time()) conn = bulk_db(BULK_DB_PATH) cur = conn.cursor() cur.execute( """ INSERT INTO bulk_jobs ( id, created_at_ts, publish_at_ts, next_attempt_ts, status, attempts, last_error, access_token, account_alias, title, privacy_level, allow_comment, allow_duet, allow_stitch, commercial_toggle, brand_organic_toggle, brand_content_toggle, is_aigc, stored_filename, original_filename, video_url, publish_id, last_status, last_fail_reason, updated_at_ts ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) """, ( job_id, now_ts, publish_at_ts, publish_at_ts, "scheduled", 0, None, access_token, account_alias or None, title, privacy_level, 1 if allow_comment else 0, 1 if allow_duet else 0, 1 if allow_stitch else 0, 1 if commercial_toggle else 0, 1 if brand_organic_toggle else 0, 1 if brand_content_toggle else 0, 1 if is_aigc else 0, stored_filename, original_fn, video_url, None, None, None, now_ts, ), ) conn.commit() conn.close() return jsonify( { "ok": True, "job_id": job_id, "publish_at_ts": publish_at_ts, "stored_filename": stored_filename, "video_url": video_url, "account_alias": account_alias or None, } ), 200 # --------------------------------------------------------------------- # Cron worker: process due scheduled jobs + refresh submitted jobs # --------------------------------------------------------------------- @app.route("/api/bulk/process_due", methods=["POST"]) def api_bulk_process_due(): require_bulk_api_key() max_jobs = int(request.form.get("max_jobs", "10") or "10") now_ts = int(time.time()) conn = bulk_db(BULK_DB_PATH) cur = conn.cursor() processed = {"scheduled_submitted": 0, "scheduled_failed": 0, "scheduled_requeued": 0, "submitted_updated": 0} # 1) scheduled due -> lock as submitting cur.execute( """ SELECT * FROM bulk_jobs WHERE status='scheduled' AND next_attempt_ts <= ? ORDER BY publish_at_ts ASC LIMIT ? """, (now_ts, max_jobs), ) due = cur.fetchall() for row in due: job_id = row["id"] # lock cur.execute( """ UPDATE bulk_jobs SET status='submitting', updated_at_ts=? WHERE id=? AND status='scheduled' """, (now_ts, job_id), ) if cur.rowcount != 1: continue conn.commit() ok, msg, _init_resp = process_one_scheduled_job( row, upload_dir=UPLOAD_DIR, public_base_url=app.config["PUBLIC_BASE_URL"], signing_secret=app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=app.config["MEDIA_TOKEN_TTL_SECONDS"], ) if ok: publish_id = msg used_access_token = resolve_access_token_for_job(row) cur.execute( """ UPDATE bulk_jobs SET status='submitted', publish_id=?, attempts=attempts+1, last_error=NULL, last_status=NULL, last_fail_reason=NULL, access_token=?, next_attempt_ts=?, updated_at_ts=? WHERE id=? AND status='submitting' """, (publish_id, used_access_token, now_ts, now_ts, job_id), ) if cur.rowcount == 1: processed["scheduled_submitted"] += 1 conn.commit() continue # failed or requeue err = msg attempts_next = int(row["attempts"]) + 1 # backoff: cannot_post_now -> requeue 5 min (hasta ~1h con 12 intentos) if err.startswith("cannot_post_now") and attempts_next <= 12: next_ts = now_ts + 300 cur.execute( """ UPDATE bulk_jobs SET status='scheduled', attempts=?, last_error=?, next_attempt_ts=?, updated_at_ts=? WHERE id=? AND status='submitting' """, (attempts_next, err, next_ts, now_ts, job_id), ) if cur.rowcount == 1: processed["scheduled_requeued"] += 1 else: cur.execute( """ UPDATE bulk_jobs SET status='failed', attempts=?, last_error=?, updated_at_ts=? WHERE id=? AND status='submitting' """, (attempts_next, err, now_ts, job_id), ) if cur.rowcount == 1: processed["scheduled_failed"] += 1 conn.commit() # 2) refresh submitted jobs statuses cur.execute( """ SELECT * FROM bulk_jobs WHERE status='submitted' ORDER BY updated_at_ts ASC LIMIT ? """, (max_jobs,), ) subs = cur.fetchall() for row in subs: job_id = row["id"] status, fail_reason = refresh_submitted_job_status(row) if not status: continue # terminal if status in ("PUBLISH_COMPLETE", "FAILED", "SEND_TO_USER_INBOX"): new_status = "complete" if status == "PUBLISH_COMPLETE" else "failed" cur.execute( """ UPDATE bulk_jobs SET status=?, last_status=?, last_fail_reason=?, updated_at_ts=? WHERE id=? AND status='submitted' """, (new_status, status, fail_reason, now_ts, job_id), ) else: cur.execute( """ UPDATE bulk_jobs SET last_status=?, last_fail_reason=?, updated_at_ts=? WHERE id=? AND status='submitted' """, (status, fail_reason, now_ts, job_id), ) if cur.rowcount == 1: processed["submitted_updated"] += 1 conn.commit() conn.close() return jsonify({"ok": True, "processed": processed, "now_ts": now_ts}), 200 # --------------------------------------------------------------------- # UI flow: local upload -> post page -> publish now (manual) # --------------------------------------------------------------------- @app.route("/upload_local", methods=["POST"]) def upload_local(): access_token = require_login_access_token() if not access_token: flash("Please connect with TikTok first.", "error") return redirect("/app") if "video" not in request.files: flash("No video file provided.", "error") return redirect("/app") file = request.files["video"] if not file.filename: flash("No file selected.", "error") return redirect("/app") original_fn = safe_filename(file.filename) draft_id = uuid.uuid4().hex stored_filename = f"{draft_id}_{original_fn}" save_path = os.path.join(UPLOAD_DIR, stored_filename) file.save(save_path) save_draft( draft_id, { "file_path": save_path, "stored_filename": stored_filename, "original_filename": original_fn, "publish_id": None, "last_init_response": None, "posting_summary": None, "status": "local_uploaded", }, ) return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) @app.route("/static_preview/", methods=["GET"]) def static_preview(draft_id: str): access_token = require_login_access_token() if not access_token: return "Not authenticated", 401 draft = load_draft(draft_id) if not draft: return "Not found", 404 path = draft.get("file_path") if not path or not os.path.exists(path): return "Not found", 404 return send_file(path, mimetype="video/mp4", as_attachment=False) @app.route("/post/", methods=["GET"]) def post_to_tiktok(draft_id: str): access_token = require_login_access_token() if not access_token: flash("Please connect with TikTok first.", "error") return redirect("/app") draft = load_draft(draft_id) if not draft: flash("Draft not found.", "error") return redirect("/app") try: creator_info = query_creator_info(access_token) except Exception as e: flash(f"Cannot post right now. Please try again later. Details: {e}", "error") return redirect("/app") can_post_now, cannot_post_reason = creator_can_post_now(creator_info) max_dur = creator_info.get("max_video_post_duration_sec") dur = video_duration_seconds_ffprobe(draft["file_path"]) duration_ok = True duration_msg = None if isinstance(max_dur, int) and max_dur > 0 and dur > 0 and dur > max_dur: duration_ok = False duration_msg = "Video is %.1fs but max allowed is %ss for this creator." % (dur, max_dur) pull_url = None stored_filename = draft.get("stored_filename") if stored_filename: pull_url = build_public_media_url( stored_filename=stored_filename, public_base_url=app.config["PUBLIC_BASE_URL"], signing_secret=app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=app.config["MEDIA_TOKEN_TTL_SECONDS"], ) return render_template( "post_to_tiktok.html", draft_id=draft_id, draft=draft, creator_info=creator_info, duration_ok=duration_ok, duration_msg=duration_msg, pull_url=pull_url, can_post_now=can_post_now, cannot_post_reason=cannot_post_reason, ) @app.route("/post//publish", methods=["POST"]) def publish(draft_id: str): access_token = require_login_access_token() if not access_token: flash("Please connect with TikTok first.", "error") return redirect("/app") draft = load_draft(draft_id) if not draft: flash("Draft not found.", "error") return redirect("/app") try: creator_info = query_creator_info(access_token) except Exception as e: flash(f"Cannot post right now. Please try again later. Details: {e}", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) if request.form.get("music_confirm") != "1": flash("You must agree to TikTok’s Music Usage Confirmation before posting.", "error") return redirect(request.referrer or url_for("ui.app_home")) can_post_now, cannot_post_reason = creator_can_post_now(creator_info) if not can_post_now: flash( cannot_post_reason or "This TikTok account cannot publish at this moment. Please try again later.", "error", ) return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) privacy_level = (request.form.get("privacy_level") or "").strip() if not privacy_level: flash("Please select a Privacy status before publishing.", "error") return redirect(request.referrer or url_for("ui.app_home")) title = (request.form.get("title") or "").strip() commercial_toggle = request.form.get("commercial_toggle") == "on" your_brand = request.form.get("your_brand") == "on" branded_content = request.form.get("branded_content") == "on" if branded_content and privacy_level == "SELF_ONLY": flash("Branded content cannot be posted with privacy set to Only me (SELF_ONLY).", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) options = creator_info.get("privacy_level_options") or [] if privacy_level not in options: flash("You must select a valid privacy option.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) allow_comment = request.form.get("allow_comment") == "on" allow_duet = request.form.get("allow_duet") == "on" allow_stitch = request.form.get("allow_stitch") == "on" if creator_info.get("comment_disabled") is True: allow_comment = False if creator_info.get("duet_disabled") is True: allow_duet = False if creator_info.get("stitch_disabled") is True: allow_stitch = False disable_comment = not allow_comment disable_duet = not allow_duet disable_stitch = not allow_stitch brand_organic_toggle = False brand_content_toggle = False if commercial_toggle: if not (your_brand or branded_content): flash( "If commercial content disclosure is on, you must select Your brand, Branded content, or both.", "error", ) return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) brand_organic_toggle = your_brand brand_content_toggle = branded_content if branded_content and privacy_level == "SELF_ONLY": flash("Branded content visibility cannot be set to private/only me.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) max_dur = creator_info.get("max_video_post_duration_sec") dur = video_duration_seconds_ffprobe(draft["file_path"]) if isinstance(max_dur, int) and max_dur > 0 and dur > 0 and dur > max_dur: flash("Video duration %.1fs exceeds allowed maximum %ss." % (dur, max_dur), "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) stored_filename = draft.get("stored_filename") if not stored_filename: flash("Internal error: missing stored_filename for draft.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) local_path = os.path.join(UPLOAD_DIR, stored_filename) if not os.path.exists(local_path): flash("Video file not found on server.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) video_url = build_public_media_url( stored_filename=stored_filename, public_base_url=app.config["PUBLIC_BASE_URL"], signing_secret=app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=app.config["MEDIA_TOKEN_TTL_SECONDS"], ) try: init_resp = upload_video_direct_post( access_token=access_token, caption=title, privacy_level=privacy_level, disable_comment=disable_comment, disable_duet=disable_duet, disable_stitch=disable_stitch, brand_content_toggle=brand_content_toggle, brand_organic_toggle=brand_organic_toggle, is_aigc=True, mode="PULL_FROM_URL", video_url=video_url, ) publish_id = (init_resp.get("data") or {}).get("publish_id") save_draft( draft_id, { **draft, "last_init_response": init_resp, "publish_id": publish_id, "posting_summary": { "title": title, "privacy_level": privacy_level, "allow_comment": allow_comment, "allow_duet": allow_duet, "allow_stitch": allow_stitch, "commercial_toggle": commercial_toggle, "your_brand": your_brand, "branded_content": branded_content, }, "status": "posted", "video_url": video_url, }, ) return redirect(url_for("ui.status_page", draft_id=draft_id)) except Exception as e: flash(f"Error posting to TikTok: {e}", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) @app.route("/status/", methods=["GET"]) def status_page(draft_id: str): access_token = require_login_access_token() if not access_token: flash("Please connect with TikTok first.", "error") return redirect("/app") draft = load_draft(draft_id) if not draft: flash("Draft not found.", "error") return redirect("/app") return render_template("status.html", draft_id=draft_id, draft=draft) @app.route("/api/status/", methods=["GET"]) def api_status(draft_id: str): access_token = require_login_access_token() if not access_token: return jsonify({"ok": False, "error": "not_authenticated"}), 401 draft = load_draft(draft_id) if not draft: return jsonify({"ok": False, "error": "draft_not_found"}), 404 publish_id = draft.get("publish_id") if not publish_id: return jsonify({"ok": True, "state": "no_publish_id_yet"}), 200 try: data = fetch_post_status(access_token, publish_id) return jsonify({"ok": True, "data": data}), 200 except Exception as e: return jsonify({"ok": False, "error": str(e)}), 200 @app.get("/api/media_url/") def api_media_url(draft_id: str): access_token = require_login_access_token() if not access_token: return jsonify({"ok": False, "error": "not_authenticated"}), 401 draft = load_draft(draft_id) if not draft: return jsonify({"ok": False, "error": "draft_not_found"}), 404 stored_filename = draft.get("stored_filename") if not stored_filename: return jsonify({"ok": False, "error": "missing_stored_filename"}), 500 local_path = os.path.join(UPLOAD_DIR, stored_filename) if not os.path.exists(local_path): return jsonify({"ok": False, "error": "file_not_found"}), 404 url = build_public_media_url( stored_filename=stored_filename, public_base_url=app.config["PUBLIC_BASE_URL"], signing_secret=app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=app.config["MEDIA_TOKEN_TTL_SECONDS"], ) return jsonify({"ok": True, "url": url}) @app.get("/healthz") def healthz(): return jsonify({"ok": True}), 200 if __name__ == "__main__": app.run(host="0.0.0.0", port=8777, debug=False)