# app/blueprints/ui.py import os import time import uuid import sqlite3 from flask import ( Blueprint, current_app, render_template, request, redirect, url_for, flash, send_file, jsonify, session, ) from db.bulk import bulk_db from services.auth import 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 token_store import load_token, save_token from tiktok_client import query_creator_info, upload_video_direct_post, fetch_post_status bp = Blueprint("ui", __name__) def _is_connected() -> bool: return session.get("tiktok_token") is not None def _ensure_alias_token_persisted(alias: str) -> None: """ Defensa extra para el caso en que ya existan sesiones antiguas donde oauth_alias estaba seteado pero token_store no lo tenía aún. """ a = (alias or "").strip() if not a: return try: if not load_token(a): tok = session.get("tiktok_token") if tok: save_token(a, tok) except Exception: # No bloqueamos UI por fallos de persistencia; el worker te lo dirá si falta alias pass # --------------------------------------------------------------------- # Pages # --------------------------------------------------------------------- @bp.route("/", endpoint="landing") def landing(): return render_template("landing.html", is_connected=_is_connected()) @bp.route("/app", endpoint="app_home") def app_home(): return render_template("index.html", is_connected=_is_connected()) @bp.route("/contact", endpoint="contact") def contact(): return render_template("contact.html", is_connected=_is_connected()) @bp.route("/tos", endpoint="tos") def tos(): return render_template("tos.html") @bp.route("/privacy", endpoint="privacy") def privacy(): return render_template("privacy.html") # --------------------------------------------------------------------- # UI flow: local upload -> post page # --------------------------------------------------------------------- @bp.route("/upload_local", methods=["POST"], endpoint="upload_local") 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") upload_dir = current_app.config["UPLOAD_DIR"] 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", "video_url": None, "bulk_job_id": None, }, ) return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) @bp.route("/static_preview/", methods=["GET"], endpoint="static_preview") 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) @bp.route("/post/", methods=["GET"], endpoint="post_to_tiktok") 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) # Nota: este pull_url es solo “preview”; caduca según TTL. 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=current_app.config["PUBLIC_BASE_URL"], signing_secret=current_app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=current_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, ) # --------------------------------------------------------------------- # Publish now (manual) # --------------------------------------------------------------------- @bp.route("/post//publish", methods=["POST"], endpoint="publish") 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 brand_content_toggle 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)) 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)) upload_dir = current_app.config["UPLOAD_DIR"] 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=current_app.config["PUBLIC_BASE_URL"], signing_secret=current_app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=current_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)) # --------------------------------------------------------------------- # Schedule from UI (creates a bulk job) # Requires post_to_tiktok.html to include publish_at (ISO with offset). # --------------------------------------------------------------------- @bp.route("/post//schedule", methods=["POST"], endpoint="schedule_post") def schedule_post(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") # Alias persistido por OAuth login/callback. El worker lo usará para refresh. account_alias = (session.get("oauth_alias") or "").strip() if not account_alias: # fallback por compat; si no existe, el worker no podrá refrescar. account_alias = "default" session["oauth_alias"] = account_alias session.modified = True _ensure_alias_token_persisted(account_alias) publish_at = (request.form.get("publish_at") or "").strip() if not publish_at: flash("Please select a schedule date/time.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) try: publish_at_ts = parse_publish_at_iso_to_ts(publish_at) except Exception as e: flash(f"Invalid publish_at (must include timezone offset). 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 scheduling.", "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 scheduling.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) 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 commercial_toggle and 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 = bool(commercial_toggle and your_brand) brand_content_toggle = bool(commercial_toggle and branded_content) if brand_content_toggle and privacy_level == "SELF_ONLY": flash("Branded content cannot be scheduled with privacy SELF_ONLY.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) # Defaults de interacción (el worker re-validará contra creator settings) allow_comment = request.form.get("allow_comment") == "on" allow_duet = request.form.get("allow_duet") == "on" allow_stitch = request.form.get("allow_stitch") == "on" # Duration enforcement (best-effort) file_path = draft.get("file_path") if not file_path or not os.path.exists(file_path): flash("Video file not found on server.", "error") return redirect(url_for("ui.post_to_tiktok", draft_id=draft_id)) stored_filename = draft.get("stored_filename") original_filename = (draft.get("original_filename") or "").strip() or os.path.basename(file_path) 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)) # Validación “UX”: si falla TikTok aquí, seguimos y dejamos que el worker lo gestione. try: creator_info = query_creator_info(access_token) options = creator_info.get("privacy_level_options") or [] if options and 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)) 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 max_dur = creator_info.get("max_video_post_duration_sec") dur = video_duration_seconds_ffprobe(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)) 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.") + " Your job was scheduled anyway; the worker will retry automatically.", "warning", ) except Exception as e: flash(f"Could not validate creator settings right now (scheduled anyway). Details: {e}", "warning") now_ts = int(time.time()) job_id = uuid.uuid4().hex # Guardamos video_url vacío para forzar “just-in-time signing” en el worker: # en bulk_worker: row['video_url'] or build_public_media_url(...) job = { "id": job_id, "created_at_ts": now_ts, "publish_at_ts": publish_at_ts, "next_attempt_ts": publish_at_ts, "status": "scheduled", "attempts": 0, "last_error": None, "access_token": access_token, # compat (worker lo usa si no hay alias) "account_alias": account_alias, # clave real para refresh en worker "title": title, "privacy_level": privacy_level, "allow_comment": 1 if allow_comment else 0, "allow_duet": 1 if allow_duet else 0, "allow_stitch": 1 if allow_stitch else 0, "commercial_toggle": 1 if commercial_toggle else 0, "brand_organic_toggle": 1 if brand_organic_toggle else 0, "brand_content_toggle": 1 if brand_content_toggle else 0, "is_aigc": 1, "stored_filename": stored_filename, "original_filename": original_filename, "video_url": "", # falsy => el worker firmará en el momento "publish_id": None, "last_status": None, "last_fail_reason": None, "updated_at_ts": now_ts, } conn = bulk_db(current_app.config["BULK_DB_PATH"]) conn.row_factory = sqlite3.Row cur = conn.cursor() # INSERT robusto (evita “N values for M columns” si tu schema cambia) cur.execute("PRAGMA table_info(bulk_jobs)") cols = [r["name"] for r in cur.fetchall()] colset = set(cols) job = {k: v for k, v in job.items() if k in colset} colnames = ", ".join(job.keys()) placeholders = ", ".join(["?"] * len(job)) cur.execute(f"INSERT INTO bulk_jobs ({colnames}) VALUES ({placeholders})", tuple(job.values())) conn.commit() conn.close() save_draft( draft_id, { **draft, "status": "scheduled", "bulk_job_id": job_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, "publish_at": publish_at, "account_alias": account_alias, }, }, ) flash(f"Scheduled job created: {job_id}", "success") return redirect(url_for("ui.bulk_job_page", job_id=job_id)) # --------------------------------------------------------------------- # Draft status page + polling # --------------------------------------------------------------------- @bp.route("/status/", methods=["GET"], endpoint="status_page") 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) @bp.route("/api/status/", methods=["GET"], endpoint="api_status") 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 @bp.get("/api/media_url/", endpoint="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 upload_dir = current_app.config["UPLOAD_DIR"] 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=current_app.config["PUBLIC_BASE_URL"], signing_secret=current_app.config["MEDIA_SIGNING_SECRET"], ttl_seconds=current_app.config["MEDIA_TOKEN_TTL_SECONDS"], ) return jsonify({"ok": True, "url": url}) # --------------------------------------------------------------------- # Bulk job UI (recommended to test cron/worker) # --------------------------------------------------------------------- @bp.get("/bulk/job/", endpoint="bulk_job_page") def bulk_job_page(job_id: str): access_token = require_login_access_token() if not access_token: flash("Please connect with TikTok first.", "error") return redirect("/app") return render_template("bulk_job.html", job_id=job_id) @bp.get("/api/bulk/job/", endpoint="api_bulk_job") def api_bulk_job(job_id: str): access_token = require_login_access_token() if not access_token: return jsonify({"ok": False, "error": "not_authenticated"}), 401 conn = bulk_db(current_app.config["BULK_DB_PATH"]) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute("SELECT * FROM bulk_jobs WHERE id=?", (job_id,)) row = cur.fetchone() conn.close() if not row: return jsonify({"ok": False, "error": "not_found"}), 404 return jsonify({"ok": True, "job": dict(row)}), 200