feat: Add TikTok OAuth auth and video publishing

- Add tiktok/ module with auth.py and publisher.py
- Implement TikTok Login Kit OAuth 2.0 flow
- Use Content Posting API for video uploads (init → upload → check status)
- Register TikTok in factory.py
- Update app.py callback for tiktok_ state prefix
- Add tiktok_client_key and tiktok_client_secret to config.py
- Update frontend: TikTok status 'Active'
- Remove old tiktok_publisher.py skeleton

social_media_publishers/app.py

@@ -76,6 +76,8 @@ def oauth2callback():
         platform = 'instagram'
     elif state.startswith('facebook_'):
         platform = 'facebook'
+    elif state.startswith('tiktok_'):
+        platform = 'tiktok'
     else:
         platform = 'youtube'  # Default for backward compatibility
     

social_media_publishers/factory.py

@@ -1,6 +1,7 @@
 from .youtube.auth import YoutubeAuthCreator
 from .instagram.auth import InstagramAuthCreator
 from .facebook.auth import FacebookAuthCreator
+from .tiktok.auth import TikTokAuthCreator
 
 class PublisherFactory:
     """Factory to create social media auth creators and publishers."""
@@ -14,6 +15,8 @@ class PublisherFactory:
             return InstagramAuthCreator()
         if platform_lower == 'facebook':
             return FacebookAuthCreator()
+        if platform_lower == 'tiktok':
+            return TikTokAuthCreator()
         raise ValueError(f"Unknown platform: {platform}")
 
     @staticmethod
@@ -28,5 +31,9 @@ class PublisherFactory:
         if platform_lower == 'facebook':
             from .facebook.publisher import FacebookPublisher
             return FacebookPublisher()
+        if platform_lower == 'tiktok':
+            from .tiktok.publisher import TikTokPublisher
+            return TikTokPublisher()
         raise ValueError(f"Unknown platform: {platform}")
 
+

social_media_publishers/frontend/src/pages/Dashboard.jsx

@@ -19,7 +19,7 @@ const platforms = [
         icon: Music2,
         color: 'text-teal-400',
         bg: 'bg-teal-400/10',
-        status: 'Coming Soon'
+        status: 'Active'
     },
     {
         id: 'instagram',

social_media_publishers/tiktok/__init__.py

@@ -0,0 +1 @@
+# TikTok auth and publisher module

social_media_publishers/tiktok/auth.py

@@ -0,0 +1,166 @@
+"""
+TikTok OAuth Authentication Handler
+Uses TikTok Login Kit (OAuth 2.0) for Content Posting API access
+"""
+
+import os
+import json
+import requests
+from typing import Dict, Any
+from ..base import SocialAuthCreator
+from src.config import get_config_value, load_configuration
+
+# Initialize Config
+load_configuration()
+
+
+class TikTokAuthCreator(SocialAuthCreator):
+    """
+    Handles TikTok OAuth 2.0 Flow via TikTok Login Kit.
+    
+    Required Environment Variables:
+        TIKTOK_CLIENT_KEY: TikTok App Client Key
+        TIKTOK_CLIENT_SECRET: TikTok App Client Secret
+    """
+    
+    # TikTok API Scopes for Content Posting
+    SCOPES = [
+        'user.info.basic',
+        'video.publish',
+        'video.upload'
+    ]
+    
+    # TikTok OAuth endpoints
+    AUTH_URL = "https://www.tiktok.com/v2/auth/authorize/"
+    TOKEN_URL = "https://open.tiktokapis.com/v2/oauth/token/"
+    USER_INFO_URL = "https://open.tiktokapis.com/v2/user/info/"
+
+    def __init__(self):
+        super().__init__(platform_name='tiktok')
+        self.client_key = get_config_value('tiktok_client_key')
+        self.client_secret = get_config_value('tiktok_client_secret')
+
+    def _get_client_config(self) -> Dict[str, Any]:
+        """Get client configuration from env vars."""
+        if self.client_key and self.client_secret:
+            return {
+                "client_key": self.client_key,
+                "client_secret": self.client_secret
+            }
+        return None
+
+    def get_auth_url(self, redirect_uri: str) -> tuple[str, str]:
+        """Generate TikTok OAuth authorization URL."""
+        config = self._get_client_config()
+        if not config:
+            raise ValueError("TikTok Client Key and Secret not configured. Set TIKTOK_CLIENT_KEY and TIKTOK_CLIENT_SECRET.")
+
+        import secrets
+        state = secrets.token_urlsafe(32)
+        
+        # TikTok requires CSRF state and code_verifier for PKCE
+        code_verifier = secrets.token_urlsafe(64)[:64]
+        
+        params = {
+            'client_key': config['client_key'],
+            'redirect_uri': redirect_uri,
+            'scope': ','.join(self.SCOPES),
+            'response_type': 'code',
+            'state': f"tiktok_{state}"
+        }
+        
+        auth_url = f"{self.AUTH_URL}?" + "&".join(f"{k}={v}" for k, v in params.items())
+        
+        # Store code_verifier in state for later use (would need session storage in practice)
+        return auth_url, f"tiktok_{state}"
+
+    def handle_callback(self, request_url: str, state: str, redirect_uri: str) -> Dict[str, Any]:
+        """Exchange authorization code for access token and save credentials."""
+        try:
+            config = self._get_client_config()
+            if not config:
+                return {"success": False, "error": "TikTok not configured"}
+
+            # Extract code from URL
+            from urllib.parse import urlparse, parse_qs
+            parsed = urlparse(request_url)
+            params = parse_qs(parsed.query)
+            code = params.get('code', [None])[0]
+            
+            if not code:
+                error = params.get('error_description', params.get('error', ['Unknown error']))[0]
+                return {"success": False, "error": error}
+
+            # Exchange code for access token
+            token_data = {
+                'client_key': config['client_key'],
+                'client_secret': config['client_secret'],
+                'code': code,
+                'grant_type': 'authorization_code',
+                'redirect_uri': redirect_uri
+            }
+            
+            headers = {'Content-Type': 'application/x-www-form-urlencoded'}
+            response = requests.post(self.TOKEN_URL, data=token_data, headers=headers)
+            
+            if response.status_code != 200:
+                return {"success": False, "error": f"Token exchange failed: {response.text}"}
+            
+            result = response.json()
+            
+            if 'error' in result:
+                return {"success": False, "error": result.get('error_description', result.get('error'))}
+            
+            access_token = result.get('access_token')
+            refresh_token = result.get('refresh_token')
+            open_id = result.get('open_id')
+            expires_in = result.get('expires_in')
+            
+            if not access_token:
+                return {"success": False, "error": "No access token returned"}
+
+            # Get user info for display name
+            username = "unknown"
+            try:
+                user_response = requests.get(
+                    self.USER_INFO_URL,
+                    headers={'Authorization': f'Bearer {access_token}'},
+                    params={'fields': 'open_id,union_id,avatar_url,display_name'}
+                )
+                if user_response.status_code == 200:
+                    user_data = user_response.json()
+                    if 'data' in user_data and 'user' in user_data['data']:
+                        username = user_data['data']['user'].get('display_name', open_id[:10])
+            except Exception as e:
+                print(f"⚠️ Could not fetch user info: {e}")
+                username = open_id[:10] if open_id else "unknown"
+
+            # Create safe filename
+            safe_name = "".join(c for c in username if c.isalnum() or c in ('-', '_')).strip()[:30]
+            if not safe_name:
+                safe_name = open_id[:20] if open_id else "default"
+            filename = f"tiktok_token_{safe_name}.json"
+            
+            token_to_save = {
+                'access_token': access_token,
+                'refresh_token': refresh_token,
+                'open_id': open_id,
+                'expires_in': expires_in,
+                'username': username
+            }
+
+            # Save using Base class method
+            self.save_token_data(token_to_save, filename)
+            
+            print(f"✅ TikTok account connected: @{username}")
+            
+            return {
+                "success": True,
+                "channel_name": f"@{username}",
+                "filename": filename
+            }
+            
+        except Exception as e:
+            import traceback
+            traceback.print_exc()
+            return {"success": False, "error": str(e)}

social_media_publishers/{tiktok_publisher.py → tiktok/publisher.py}

@@ -1,105 +1,179 @@
 """
-TikTok Video Scheduler for GitHub Actions
-Uses environment variables for authentication + APIClients for GCS download
+TikTok Video Publisher
+Uses TikTok Content Posting API for publishing videos
 """
 
-import asyncio
-import requests
-import json
 import os
 import sys
-import base64
+import json
 import time
-import pandas as pd
-from datetime import datetime
+import requests
+from typing import Dict, Any, Optional
+
+# Add parent directory to path
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))
 
-# Add parent directory to path to allow importing modules from src
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
+from social_media_publishers.base import SocialPublisher
+from src.config import get_config_value, load_configuration
 
-from dotenv import load_dotenv
-from pathlib import Path
-from asset_manager.content_strategy_lib import get_content_strategy_lib
-from config import get_config_value
-import hashlib
-from google_src.gcs_utils import find_and_download_gcs_file
+# Initialize Config
+load_configuration()
 
-DATA_DIR = Path("data")
 
-class TikTokPublisher:
+class TikTokPublisher(SocialPublisher):
+    """
+    Publishes videos to TikTok via Content Posting API.
+    
+    Uses TokenManager for encrypted credential storage.
+    """
+    
+    BASE_URL = "https://open.tiktokapis.com/v2"
+    
     def __init__(self):
+        """Initialize TikTok publisher."""
+        self.access_token = None
+        self.open_id = None
+        self.username = None
+        self.token_manager = None
+        
+        # Try to initialize TokenManager
+        encryption_key = get_config_value("encryption_key")
+        hf_token = get_config_value("hf_token")
+        
+        if encryption_key and hf_token:
+            try:
+                from youtube_auto_pub.token_manager import TokenManager
+                from youtube_auto_pub.config import YouTubeConfig
+                
+                config = YouTubeConfig(
+                    encryption_key=encryption_key,
+                    hf_token=hf_token,
+                    hf_repo_id="ai-workbench/data_process"
+                )
+                self.token_manager = TokenManager(config)
+            except Exception as e:
+                print(f"⚠️ Could not initialize TokenManager: {e}")
+        
+        # Note: Authentication happens when publish() is called via app.py
+        # The account is passed at runtime, not from env config
+
+    def authenticate(self, account_id: Optional[str] = None) -> None:
         """
-        Initialize TikTok scheduler using environment variables
+        Authenticate with TikTok using stored credentials.
         
-        Required Environment Variables:
-            TIKTOK_ACCESS_TOKEN: TikTok API access token
+        Args:
+            account_id: TikTok username/display name for TokenManager lookup
         """
-        load_dotenv()
-        self.access_token = get_config_value('tiktok_access_token')
-        self.base_url = "https://open.tiktokapis.com/v2"
+        if not self.token_manager:
+            raise ValueError("TokenManager not initialized (check ENCRYPTION_KEY/HF_TOKEN)")
         
-        if not self.access_token:
-            raise ValueError("❌ Missing required environment variable: TIKTOK_ACCESS_TOKEN must be set")
+        # account_id is required - passed from UI/API, not from env
+            
+        if not account_id:
+            raise ValueError("account_id (TikTok username) is required for authentication")
+
+        try:
+            print(f"🔄 Loading credentials for TikTok: @{account_id}")
+            
+            # Construct filename
+            safe_name = "".join(c for c in account_id if c.isalnum() or c in ('-', '_')).strip()[:30]
+            filename = f"tiktok_token_{safe_name}.json"
+            
+            # Download and decrypt
+            file_path = self.token_manager.download_and_decrypt(filename)
+            
+            if not os.path.exists(file_path):
+                raise FileNotFoundError(f"Token file not found: {file_path}")
+            
+            with open(file_path, 'r') as f:
+                token_data = json.load(f)
+            
+            self.access_token = token_data.get('access_token')
+            self.open_id = token_data.get('open_id')
+            self.username = token_data.get('username')
+            
+            if not self.access_token:
+                raise ValueError("Invalid token data - missing access_token")
+            
+            print(f"✅ Authenticated as @{self.username}")
+            
+        except Exception as e:
+            print(f"❌ Authentication failed: {e}")
+            raise
 
-    def upload_video(self, video_path, title, privacy_level=None,
-                     disable_duet=False, disable_comment=False, disable_stitch=False):
+    def publish(self, content_path: str, metadata: Dict[str, Any]) -> Dict[str, Any]:
         """
-        Upload a video to TikTok
+        Publish a video to TikTok.
         
         Args:
-            video_path: Local path to video file or base64 encoded string
-            title: Video title/caption (max 2200 chars)
-            privacy_level: PUBLIC_TO_EVERYONE, MUTUAL_FOLLOW_FRIENDS, SELF_ONLY
-            disable_duet: Disable duet (default False)
-            disable_comment: Disable comments (default False)
-            disable_stitch: Disable stitch (default False)
+            content_path: Path to video file (or GCS path)
+            metadata: Dict with 'title', 'privacy_level', etc.
         
         Returns:
-            dict: Response with publish_id or error
+            Dict with success status and publish_id or error
         """
-        if privacy_level is None:
-            privacy_level = get_config_value('tiktok_privacy_level', 'PUBLIC_TO_EVERYONE')
-
-        if not os.path.exists(video_path):
-            return {'error': f'Video file not found: {video_path}'}
+        if not self.access_token:
+            raise ValueError("Not authenticated. Call authenticate() first.")
+        
+        # Prepare content (handles GCS download if needed)
+        local_path = self.prepare_content(content_path)
+        
+        if not os.path.exists(local_path):
+            return {"error": f"Video file not found: {local_path}"}
         
+        title = metadata.get('title', metadata.get('description', metadata.get('caption', '')))
+        privacy_level = metadata.get('privacy_level', 'PUBLIC_TO_EVERYONE')
+        disable_duet = metadata.get('disable_duet', False)
+        disable_comment = metadata.get('disable_comment', False)
+        disable_stitch = metadata.get('disable_stitch', False)
+        
+        return self._upload_video(local_path, title, privacy_level, 
+                                  disable_duet, disable_comment, disable_stitch)
+
+    def _upload_video(self, video_path: str, title: str, privacy_level: str,
+                      disable_duet: bool, disable_comment: bool, disable_stitch: bool) -> Dict[str, Any]:
+        """
+        Upload video to TikTok using the Content Posting API.
+        3-step process: Initialize → Upload → Check Status
+        """
         video_size = os.path.getsize(video_path)
-        print(f"🎬 Initializing upload for: {title[:50]} ({video_size / 1024:.2f} KB)")
+        print(f"🎬 Uploading TikTok video: {title[:50]}... ({video_size / 1024 / 1024:.1f} MB)")
         
-        # Step 1: Initialize
-        init_response = self._initialize_upload(video_size, title, privacy_level,
-                                                disable_duet, disable_comment, disable_stitch)
-        if 'error' in init_response:
-            print(f"❌ Initialization failed: {init_response['error']}")
-            return init_response
+        # Step 1: Initialize upload
+        init_result = self._initialize_upload(video_size, title, privacy_level,
+                                               disable_duet, disable_comment, disable_stitch)
+        if 'error' in init_result:
+            return init_result
         
-        publish_id = init_response['publish_id']
-        upload_url = init_response['upload_url']
-        print(f"✅ Upload initialized → Publish ID: {publish_id}")
+        publish_id = init_result['publish_id']
+        upload_url = init_result['upload_url']
+        print(f"✅ Upload initialized: {publish_id}")
         
-        # Step 2: Upload file
+        # Step 2: Upload video file
         upload_success = self._upload_file(upload_url, video_path)
         if not upload_success:
-            print("❌ Video upload failed")
-            return {'error': 'Video upload failed'}
+            return {"error": "Video upload failed"}
+        
+        print("✅ Video uploaded, checking status...")
         
-        print("✅ Video uploaded successfully, finalizing...")
         # Step 3: Check publish status
         return self._check_publish_status(publish_id)
-    
-    def _initialize_upload(self, video_size, title, privacy_level,
-                           disable_duet, disable_comment, disable_stitch):
-        """Initialize the upload process"""
-        endpoint = f"{self.base_url}/post/publish/video/init/"
+
+    def _initialize_upload(self, video_size: int, title: str, privacy_level: str,
+                           disable_duet: bool, disable_comment: bool, disable_stitch: bool) -> Dict[str, Any]:
+        """Initialize the upload process."""
+        endpoint = f"{self.BASE_URL}/post/publish/video/init/"
         headers = {
             'Authorization': f'Bearer {self.access_token}',
             'Content-Type': 'application/json; charset=UTF-8'
         }
-        chunk_size = min(video_size, 64 * 1024 * 1024)
+        
+        chunk_size = min(video_size, 64 * 1024 * 1024)  # Max 64MB chunks
         total_chunks = max(1, (video_size + chunk_size - 1) // chunk_size)
         
         data = {
             "post_info": {
-                "title": title,
+                "title": title[:2200],  # TikTok limit
                 "privacy_level": privacy_level,
                 "disable_duet": disable_duet,
                 "disable_comment": disable_comment,
@@ -115,35 +189,44 @@ class TikTokPublisher:
 
         response = requests.post(endpoint, headers=headers, json=data)
         result = response.json()
+        
         if response.status_code == 200 and 'data' in result:
             return {
                 'publish_id': result['data']['publish_id'],
                 'upload_url': result['data']['upload_url']
             }
+        
         error_msg = result.get('error', {}).get('message', 'Initialization failed')
         return {'error': error_msg, 'full_response': result}
-    
-    def _upload_file(self, upload_url, video_path):
-        """Upload the video file to TikTok's upload URL"""
+
+    def _upload_file(self, upload_url: str, video_path: str) -> bool:
+        """Upload the video file to TikTok's upload URL."""
         try:
             with open(video_path, 'rb') as f:
                 video_data = f.read()
+            
+            file_size = len(video_data)
             headers = {
                 'Content-Type': 'video/mp4',
-                'Content-Length': str(len(video_data))
+                'Content-Length': str(file_size),
+                'Content-Range': f'bytes 0-{file_size-1}/{file_size}'
             }
+            
             response = requests.put(upload_url, headers=headers, data=video_data)
-            if response.status_code in [200, 204]:
+            
+            if response.status_code in [200, 201, 204]:
                 return True
-            print(f"⚠️ Upload failed: {response.text}")
+            
+            print(f"⚠️ Upload failed ({response.status_code}): {response.text}")
             return False
+            
         except Exception as e:
             print(f"❌ Upload exception: {e}")
             return False
-    
-    def _check_publish_status(self, publish_id, max_attempts=15):
-        """Poll TikTok API to check upload/publish status"""
-        endpoint = f"{self.base_url}/post/publish/status/fetch/"
+
+    def _check_publish_status(self, publish_id: str, max_attempts: int = 20) -> Dict[str, Any]:
+        """Poll TikTok API to check upload/publish status."""
+        endpoint = f"{self.BASE_URL}/post/publish/status/fetch/"
         headers = {
             'Authorization': f'Bearer {self.access_token}',
             'Content-Type': 'application/json; charset=UTF-8'
@@ -151,6 +234,7 @@ class TikTokPublisher:
         data = {"publish_id": publish_id}
         
         print("⏳ Checking publish status...")
+        
         for attempt in range(1, max_attempts + 1):
             response = requests.post(endpoint, headers=headers, json=data)
             result = response.json()
@@ -158,11 +242,13 @@ class TikTokPublisher:
             if response.status_code == 200 and 'data' in result:
                 status = result['data'].get('status')
                 print(f"  Attempt {attempt}/{max_attempts}: {status}")
+                
                 if status == 'PUBLISH_COMPLETE':
                     print("✅ Video published successfully!")
                     return {
                         'success': True,
                         'publish_id': publish_id,
+                        'video_id': publish_id,
                         'status': status
                     }
                 elif status == 'FAILED':
@@ -173,108 +259,5 @@ class TikTokPublisher:
                 print(f"⚠️ Status check error: {error_msg}")
             
             time.sleep(10)
-        return {'error': 'Publishing timeout - still processing'}
-
-# ===========================================================
-# MAIN (async) — CSV-driven + GCS download like YouTubePublisher
-# ===========================================================
-async def main():
-    try:
-        scheduler = TikTokPublisher()
         
-        content_lib = get_content_strategy_lib()
-        df = content_lib.strategies
-        
-        all_rows = []
-        worksheet_name = get_config_value("content_strategy_worksheet", "TikTok_Upload")
-        for i, row in df.iterrows():
-            all_rows.append((worksheet_name, row.to_dict()))
-        
-        print(f"📈 Found {len(all_rows)} TikTok videos to upload.")
-        
-        for idx, (csv_name, row) in enumerate(all_rows):
-            tts_script = row.get("TTS Script (AI Avatar)", "").strip()
-            title = row.get("Captions", "").strip()
-            
-            print("\n" + "="*50)
-            print(f"🎞️ Uploading TikTok {idx + 1}/{len(all_rows)}")
-            print(f"🎵 Title: {title[:80]}")
-            print("="*50)
-            
-            # Download from GCS
-            local_path = find_and_download_gcs_file(tts_script)
-            if not local_path or not os.path.exists(local_path):
-                print(f"❌ File not found, skipping: {tts_script}")
-                continue
-            
-            result = scheduler.upload_video(video_path=local_path, title=title)
-            print("\nRESULT:")
-            print(json.dumps(result, indent=2))
-            print("="*50)
-            
-            if not result.get('success'):
-                sys.exit(1)
-        
-        print("✅ All TikTok uploads complete!")
-        sys.exit(0)
-    
-    except Exception as e:
-        print(f"❌ Fatal error: {e}")
-        import traceback
-        traceback.print_exc()
-        sys.exit(1)
-
-if __name__ == "__main__":
-    asyncio.run(main())
-
-"""
-GitHub Actions Workflow Example:
-================================
-
-name: Post to TikTok
-on:
-  schedule:
-    - cron: '0 14 * * *'  # Daily at 2 PM UTC
-  workflow_dispatch:
-    inputs:
-      video_path:
-        description: 'Video file path'
-        required: true
-      title:
-        description: 'Video title'
-        required: true
-
-jobs:
-  post:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      
-      - name: Set up Python
-        uses: actions/setup-python@v4
-        with:
-          python-version: '3.10'
-      
-      - name: Install dependencies
-        run: pip install requests
-      
-      - name: Post to TikTok
-        env:
-          TIKTOK_ACCESS_TOKEN: ${{ secrets.TIKTOK_ACCESS_TOKEN }}
-          VIDEO_PATH: ${{ github.event.inputs.video_path || 'videos/my-video.mp4' }}
-          VIDEO_TITLE: ${{ github.event.inputs.title || 'Check this out! #fyp #viral' }}
-          TIKTOK_PRIVACY_LEVEL: 'PUBLIC_TO_EVERYONE'
-          DISABLE_DUET: 'false'
-          DISABLE_COMMENT: 'false'
-          DISABLE_STITCH: 'false'
-        run: python tiktok_scheduler.py
-
-Required GitHub Secrets:
-- TIKTOK_ACCESS_TOKEN
-
-Optional Environment Variables:
-- TIKTOK_PRIVACY_LEVEL (default: PUBLIC_TO_EVERYONE)
-- DISABLE_DUET (default: false)
-- DISABLE_COMMENT (default: false)
-- DISABLE_STITCH (default: false)
-"""
+        return {'error': 'Publishing timeout - still processing', 'publish_id': publish_id}

src/config.py

@@ -190,6 +190,8 @@ def load_configuration(force_reload: bool = False) -> Dict[str, Any]:
         # TikTok
         "tiktok_access_token": os.getenv("TIKTOK_ACCESS_TOKEN"),
         "tiktok_privacy_level": os.getenv("TIKTOK_PRIVACY_LEVEL", "PUBLIC_TO_EVERYONE"),
+        "tiktok_client_key": os.getenv("TIKTOK_CLIENT_KEY"),
+        "tiktok_client_secret": os.getenv("TIKTOK_CLIENT_SECRET"),
         
         # YouTube
         "youtube_refresh_token": os.getenv("YOUTUBE_REFRESH_TOKEN"),