# 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 uuid import time import sqlite3 from typing import Optional, Dict, Any, Tuple from config import ( FLASK_SECRET_KEY, PUBLIC_BASE_URL, MEDIA_SIGNING_SECRET, MEDIA_TOKEN_TTL_SECONDS, ) 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 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() # --------------------------------------------------------------------- # 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") # --------------------------------------------------------------------- # 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 # --------------------------------------------------------------------- # 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 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) 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=? """, (publish_id, used_access_token, now_ts, now_ts, job_id), ) 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=? """, (attempts_next, err, next_ts, now_ts, job_id), ) processed["scheduled_requeued"] += 1 else: cur.execute( """ UPDATE bulk_jobs SET status='failed', attempts=?, last_error=?, updated_at_ts=? WHERE id=? """, (attempts_next, err, now_ts, job_id), ) 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=? """, (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=? """, (status, fail_reason, now_ts, job_id), ) 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)