Grok Media Skill

# Ignore data directory containing sensitive tokens
libs/data/*
!libs/data/*.example.json
!libs/data/*.example.toml
__pycache__/
*.pyc
.DS_Store
output/
generated_assets/

[grok]
# 必填：用于防拦截的 User-Agent，建议通过浏览器控制台 navigator.userAgent 获取
user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36"

# 选填：用于绕过 Cloudflare 的 Cookie
cf_clearance = ""

# 强烈建议：填入浏览器中的完整 Cookie 字符串 (包含 cf_clearance, sso-rw, _ga 等)
cookie = ""

# 代理配置：若浏览器使用了代理，此处必须填入相同的地址以保持 IP 一致
proxy_url = "http://127.0.0.1:7890"

# 动态 Statsig 特征，建议设为 true
dynamic_statsig = true

# 生成参数
temporary = false
show_thinking = true

[global]
log_level = "INFO"
# 图片保存模式: "url" 或 "base64"
image_mode = "url"

{
    "desc": "Put your Grok SSO token here. Get it from browser cookies (sso-rw / sso).",
    "ssoNormal": {
        "YOUR_JWT_HERE": {
            "status": "active",
            "remark": "Default Account"
        }
    },
    "ssoSuper": {
        "OPTIONAL_SUPER_JWT": {
            "status": "inactive",
            "remark": "Super Account (Optional)"
        }
    }
}

import json
import uuid
import base64
import random
import string
import os
from pathlib import Path

# 配置路径 (指向 shared_utils 平级的 data 目录，或者用户指定的目录)
# 暂时假设 data 目录在 shared_utils/data
CURRENT_DIR = Path(__file__).parent
DATA_DIR = CURRENT_DIR / "data"
TOKEN_FILE = DATA_DIR / "token.json"
SETTING_FILE = DATA_DIR / "setting.toml"

def ensure_data_dir():
    DATA_DIR.mkdir(parents=True, exist_ok=True)

def get_cf_clearance():
    if not SETTING_FILE.exists(): return ""
    try:
        import toml
        data = toml.load(SETTING_FILE)
        cf = data.get("grok", {}).get("cf_clearance", "")
        if cf and not cf.startswith("cf_clearance="):
            cf = f"cf_clearance={cf}"
        return cf
    except: return ""

def _generate_statsig_id():
    import base64
    import random
    import string
    def rand_str(length, letters_only=True):
        chars = string.ascii_lowercase if letters_only else string.ascii_lowercase + string.digits
        return ''.join(random.choices(chars, k=length))
    
    if random.choice([True, False]):
        rand = rand_str(5, letters_only=False)
        msg = f"e:TypeError: Cannot read properties of null (reading 'children['{rand}']')"
    else:
        rand = rand_str(10)
        msg = f"e:TypeError: Cannot read properties of undefined (reading '{rand}')"
    return base64.b64encode(msg.encode()).decode()

def get_sso_jwt():
    if not TOKEN_FILE.exists(): return None
    try:
        with open(TOKEN_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
            for cat in ["ssoNormal", "ssoSuper"]:
                if cat in data and data[cat]:
                    for jwt, info in data[cat].items():
                        if info.get("status") == "active": return jwt
    except: return None

def get_headers(cookie_str, path="/"):
    content_type = "text/plain;charset=UTF-8" if "upload-file" in path else "application/json"
    
    # 动态获取配置
    import toml
    try:
        data = toml.load(SETTING_FILE)
        grok_cfg = data.get("grok", {})
        dynamic_statsig = grok_cfg.get("dynamic_statsig", True)
        statsig = _generate_statsig_id() if dynamic_statsig else grok_cfg.get("x_statsig_id")
        ua = grok_cfg.get("user_agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36")
    except:
        statsig = _generate_statsig_id()
        ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36"

    headers = {
        "User-Agent": ua,
        "Accept": "*/*",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
        "Content-Type": content_type,
        "x-statsig-id": statsig,
        "x-xai-request-id": str(uuid.uuid4()),
        "x-traceparent": "00-e172a5a544b6c39f993d39c06346b0a6-577660633c5e0da6-00", # Use a dummy or dynamic
        "Cookie": cookie_str,
        "Referer": "https://grok.com/",
        "Origin": "https://grok.com",
        "Priority": "u=1, i",
        "sec-ch-ua": '"Not(A:Brand";v="8", "Chromium";v="133", "Google Chrome";v="133"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin"
    }
    return headers

def get_full_cookie():
    if not SETTING_FILE.exists(): return ""
    try:
        import toml
        data = toml.load(SETTING_FILE)
        return data.get("grok", {}).get("cookie", "")
    except: return ""

def get_auth_context():
    """获取完整的认证需要的上下文 (Cookie)"""
    # 优先尝试从 setting.toml 获取完整 Cookie
    full_cookie = get_full_cookie()
    if full_cookie:
        return full_cookie
        
    jwt = get_sso_jwt()
    if not jwt:
        return None
    
    cookie = f"sso-rw={jwt}; sso={jwt}"
    cf = get_cf_clearance()
    if cf: cookie = f"{cookie}; {cf}"
    return cookie

import sys
import json
import time
import os
from pathlib import Path
from curl_cffi import requests

# Add libs to path
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))

try:
    from grok_client import get_auth_context, get_headers
except ImportError:
    print("[-] Error: libs module not found. Path:", libs_path)
    print("[-] Error: shared_utils module not found")
    sys.exit(1)

def generate_image_grok(prompt, ratio=None, count=1):
    cookie_str = get_auth_context()
    if not cookie_str:
        print("[-] Error: Authentication failed (Token not found)")
        return []

    # 强制强化提示词
    safe_prompt = f"Generate an image of {prompt}" if "generate" not in prompt.lower() else prompt
    if ratio:
        safe_prompt = f"{safe_prompt} --aspect {ratio}"
        
    # 从 setting.toml 读取代理
    from grok_client import SETTING_FILE
    proxy = None
    try:
        import toml
        data = toml.load(SETTING_FILE)
        proxy = data.get("grok", {}).get("proxy_url")
    except: pass
    proxies = {"http": proxy, "https": proxy} if proxy else None

    print(f"[*] Sending prompt to Grok: '{safe_prompt}'")
    print(f"[*] Using Proxy: {proxy}")
    
    payload = {
        "temporary": False,
        "modelName": "grok-4-1-thinking-1129",
        "message": safe_prompt, 
        "fileAttachments": [],
        "imageAttachments": [],
        "disableSearch": False,
        "enableImageGeneration": True,
        "imageGenerationCount": count,
        "modelMode": "MODEL_MODE_AUTO"
    }

    try:
        response = requests.post(
            "https://grok.com/rest/app-chat/conversations/new",
            headers=get_headers(cookie_str),
            json=payload,
            impersonate="chrome133a",
            stream=True,
            timeout=180,
            proxies=proxies
        )
        
        image_urls = []
        for line in response.iter_lines():
            if not line: continue
            try:
                line_str = line.decode('utf-8')
                data = json.loads(line_str)
                m_resp = data.get("result", {}).get("response", {}).get("modelResponse", {})
                urls = m_resp.get("generatedImageUrls", [])
                for u in urls:
                    if u not in image_urls: image_urls.append(u)
            except: continue
        
        if not image_urls:
            print("[-] No images returned.")
            return []

        print(f"[+] Generated {len(image_urls)} images. Downloading...")
        
        # Save to user's project root directory
        # current structure: <root>/.agent/skills/grok-media-skill/libs/
        project_root = libs_path.parent.parent.parent.parent 
        save_dir = project_root / "generated_assets"
        save_dir.mkdir(parents=True, exist_ok=True)
        
        local_files = []

        for i, url_path in enumerate(image_urls):
            full_url = f"https://assets.grok.com/{url_path}"
            # Create a simple name
            safe_name = f"grok_{int(time.time())}_{i}.jpg"
            local_path = save_dir / safe_name
            
            try:
                dl_headers = get_headers(cookie_str)
                if "x-statsig-id" in dl_headers:
                    del dl_headers["x-statsig-id"]
                
                r_dl = requests.get(full_url, headers=dl_headers, impersonate="chrome133a", timeout=30)
                if r_dl.status_code == 200:
                    local_path.write_bytes(r_dl.content)
                    print(f" [V] Saved: {local_path}")
                    local_files.append(str(local_path))
            except Exception as e:
                print(f" [X] Download failed: {e}")
                
        return local_files

    except Exception as e:
        print(f"[-] Request failed: {e}")
        return []

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python generate.py <prompt> [ratio]")
        sys.exit(1)
        
    p = sys.argv[1]
    r = sys.argv[2] if len(sys.argv) > 2 else None
    generate_image_grok(p, r)

import sys
import json
import time
import base64
import os
from pathlib import Path
from curl_cffi import requests

# Add libs to path
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))
# In this unified skill, generate_image is just next door
sys.path.append(str(current_dir))

try:
    from grok_client import get_auth_context, get_headers
    from generate_image import generate_image_grok
except ImportError:
    print("[-] Error: libs or generate_image module not found")
    print("[-] Error: shared_utils or text-to-image module not found")
    sys.exit(1)

def upload_image(image_input, cookie_str):
    # 获取代理
    from grok_client import SETTING_FILE
    proxy = None
    try:
        import toml
        data = toml.load(SETTING_FILE)
        proxy = data.get("grok", {}).get("proxy_url")
    except: pass
    proxies = {"http": proxy, "https": proxy} if proxy else None

    # 逻辑：如果是 URL
    if image_input.startswith("http"):
        print(f"[*] Downloading reference image...")
        try:
            r = requests.get(image_input, impersonate="chrome133a", proxies=proxies)
            if r.status_code == 200:
                mime = r.headers.get("content-type", "image/png")
                b64_content = base64.b64encode(r.content).decode()
            else:
                print(f"[-] Image download failed: {r.status_code}")
                return None, None
        except Exception as e:
             print(f"[-] Image download error: {e}")
             return None, None
    else:
        # 本地文件
        p = Path(image_input)
        if p.exists():
            print(f"[*] Reading local image: {p.name}...")
            import mimetypes
            mime, _ = mimetypes.guess_type(image_input)
            mime = mime or "image/png"
            b64_content = base64.b64encode(p.read_bytes()).decode()
        else:
            print(f"[-] Local file not found: {image_input}")
            return None, None

    print("[*] Uploading to asset library...")
    payload = {"fileName": "seed.png", "fileMimeType": mime, "content": b64_content}
    
    try:
        # 上传需要特殊的 header
        headers = get_headers(cookie_str, "/upload-file")
        resp = requests.post("https://grok.com/rest/app-chat/upload-file", 
                             headers=headers, 
                             data=json.dumps(payload), impersonate="chrome133a", proxies=proxies)
        
        if resp.status_code == 200:
            d = resp.json()
            fid = d.get("fileMetadataId")
            furi = d.get("fileUri")
            print(f"[debug] Upload success. ID: {fid}")
            return fid, furi
        else:
            print(f"[-] Upload failed ({resp.status_code})")
    except Exception as e:
        print(f"[-] Upload exception: {e}")
    return None, None

def create_post(file_uri, cookie_str):
    print("[*] Creating Media Post anchor...")
    # 获取代理
    from grok_client import SETTING_FILE
    proxy = None
    try:
        import toml
        data = toml.load(SETTING_FILE)
        proxy = data.get("grok", {}).get("proxy_url")
    except: pass
    proxies = {"http": proxy, "https": proxy} if proxy else None

    payload = {"media_url": f"https://assets.grok.com/{file_uri}", "media_type": "MEDIA_POST_TYPE_IMAGE"}
    try:
        resp = requests.post("https://grok.com/rest/media/post/create", 
                             headers=get_headers(cookie_str, "/create"), 
                             json=payload, impersonate="chrome133a", proxies=proxies)
        if resp.status_code == 200:
            pid = resp.json().get("post", {}).get("id")
            return pid
        else:
            print(f"[-] Create Post failed ({resp.status_code})")
    except Exception as e:
        print(f"[-] Create Post exception: {e}")
    return None

def run_video_gen(post_id, file_id, prompt, cookie_str):
    print(f"[*] Requesting video generation (This takes 1-2 mins)...")
    
    # 从 setting.toml 读取代理
    from grok_client import SETTING_FILE
    proxy = None
    try:
        import toml
        data = toml.load(SETTING_FILE)
        proxy = data.get("grok", {}).get("proxy_url")
    except: pass
    proxies = {"http": proxy, "https": proxy} if proxy else None

    payload = {
        "temporary": False,
        "modelName": "grok-4-1-thinking-1129",
        "message": f"https://grok.com/imagine/{post_id} {prompt} --mode=custom",
        "fileAttachments": [file_id] if file_id else [],
        "toolOverrides": {"videoGen": True},
        "modelMode": "MODEL_MODE_AUTO"
    }

    headers = get_headers(cookie_str)
    headers["Referer"] = f"https://grok.com/imagine/{post_id}"

    try:
        response = requests.post(
            "https://grok.com/rest/app-chat/conversations/new",
            headers=headers, 
            json=payload, 
            impersonate="chrome133a", 
            stream=True, 
            timeout=300,
            proxies=proxies 
        )
        
        last_p = -1
        v_url = None
        
        for line in response.iter_lines():
            if not line: continue
            try:
                line_str = line.decode('utf-8')
                if "<html" in line_str.lower():
                    print("\n[!] Cloudflare Security Challenge Detected.")
                    break
                data = json.loads(line_str)
                
                if "error" in data:
                    print(f"\n[-] API Error: {data['error']}")
                    return

                res = data.get("result", {}).get("response", {}).get("streamingVideoGenerationResponse", {})
                if res:
                    p = res.get("progress", 0)
                    url = res.get("videoUrl")
                    
                    if p > last_p: 
                        print(f" [Generating: {p}%] ", end="\r")
                        last_p = p
                    if url: v_url = url
            except: continue
        
        if v_url:
            print(f"\n[+] Video generated! URL: {v_url}")
            # current structure: <root>/.agent/skills/grok-media-skill/libs/
            project_root = libs_path.parent.parent.parent.parent
            save_dir = project_root / "generated_assets"
            save_dir.mkdir(parents=True, exist_ok=True)
            save_path = save_dir / f"video_{int(time.time())}.mp4"
            
            print(f"[*] Downloading...")
            dl_headers = get_headers(cookie_str)
            if "x-statsig-id" in dl_headers:
                del dl_headers["x-statsig-id"] 
            
            r_dl = requests.get(f"https://assets.grok.com/{v_url}", headers=dl_headers, impersonate="chrome133a", proxies=proxies)
            if r_dl.status_code == 200:
                save_path.write_bytes(r_dl.content)
                print(f"[V] Saved video: {save_path}")
                return str(save_path)
            else: print(f"[-] Download failed: {r_dl.status_code}")
        else: 
            print("\n[-] Generation ended without video URL.")
            
    except Exception as e: print(f"\n[-] Video request failed: {e}")
    return None

def main():
    if len(sys.argv) < 2:
        print("Usage: python generate_video.py <prompt> [image_path_or_ratio]")
        return

    prompt = sys.argv[1]
    arg2 = sys.argv[2] if len(sys.argv) > 2 else None
    
    image_input = None
    ratio = None
    
    def is_ratio(s):
        return s and (":" in s or "--" in s or len(s) < 10)
    
    if arg2:
        if is_ratio(arg2) and not Path(arg2).exists() and not arg2.startswith("http"):
            ratio = arg2
        else:
            image_input = arg2

    cookie = get_auth_context()
    if not cookie:
        print("[-] Error: Authentication failed")
        return

    if not image_input:
        print(f"[*] No base image provided. Generating base image first (Ratio: {ratio or 'Default'})...")
        # Reuse logic from text-to-image
        generated_paths = generate_image_grok(prompt, ratio, count=1)
        if generated_paths:
            image_input = generated_paths[0]
            print(f"[+] Base image generated: {image_input}")
        else:
            print("[-] Failed to generate base image")
            return

    # Now we have image_input, proceed to generate video
    fid, furi = upload_image(image_input, cookie)
    if fid and furi:
        pid = create_post(furi, cookie)
        if pid:
            run_video_gen(pid, fid, prompt, cookie)
        else:
            print("[-] Failed to create post anchor")
    else:
        print("[-] Failed to upload asset")

if __name__ == "__main__":
    main()

import sys
import json
import os
from pathlib import Path
from curl_cffi import requests

# Add libs to path
current_dir = Path(__file__).parent
libs_path = current_dir.parent / "libs"
sys.path.append(str(libs_path))

try:
    from grok_client import get_auth_context, get_headers
except ImportError:
    print("[-] Error: libs module not found.")
    sys.exit(1)

def test_chat(prompt="Hi"):
    cookie_str = get_auth_context()
    if not cookie_str:
        print("[-] Error: Authentication failed (Token not found)")
        return

    # 从 setting.toml 读取代理
    from grok_client import SETTING_FILE
    proxy = None
    try:
        import toml
        data = toml.load(SETTING_FILE)
        proxy = data.get("grok", {}).get("proxy_url")
    except: pass
    
    proxies = {"http": proxy, "https": proxy} if proxy else None

    print(f"[*] Testing Grok Chat with prompt: '{prompt}'")
    print(f"[*] Using Proxy: {proxy}")
    
    payload = {
        "temporary": False,
        "modelName": "grok-4-1-thinking-1129",
        "message": prompt,
        "fileAttachments": [],
        "imageAttachments": [],
        "disableSearch": False,
        "enableImageGeneration": True,
        "returnImageBytes": False,
        "returnRawGrokInXaiRequest": False,
        "enableImageStreaming": True,
        "imageGenerationCount": 2,
        "forceConcise": False,
        "toolOverrides": {},
        "enableSideBySide": True,
        "sendFinalMetadata": True,
        "customPersonality": "Respond briefly and directly, using as few words as possible.",
        "isReasoning": False,
        "disableTextFollowUps": False,
        "responseMetadata": {
            "modelConfigOverride": {"modelMap": {}},
            "requestModelDetails": {"modelId": "grok-4-1-thinking-1129"}
        },
        "disableMemory": False,
        "forceSideBySide": False,
        "modelMode": "MODEL_MODE_AUTO",
        "isAsyncChat": False,
        "deviceEnvInfo": {
            "darkModeEnabled": True,
            "devicePixelRatio": 1,
            "screenWidth": 2560,
            "screenHeight": 1440,
            "viewportWidth": 937,
            "viewportHeight": 456
        }
    }

    try:
        response = requests.post(
            "https://grok.com/rest/app-chat/conversations/new",
            headers=get_headers(cookie_str),
            json=payload,
            impersonate="chrome133a", # Use Chrome for better TLS fingerprint
            stream=True,
            timeout=120,
            proxies=proxies
        )
        
        if response.status_code != 200:
            print(f"[-] HTTP Error: {response.status_code}")
            print(f"[-] Response: {response.text[:500]}")
            return

        full_text = ""
        is_thinking = False
        print("[*] Receiving response stream...")
        
        for line in response.iter_lines():
            if not line: continue
            line_str = line.decode('utf-8')
            
            try:
                data = json.loads(line_str)
                res_obj = data.get("result", {}).get("response", {})
                
                # 检查是否在思考
                current_thinking = res_obj.get("isThinking", False)
                if current_thinking and not is_thinking:
                    print("\n[Thinking] ", end="", flush=True)
                    is_thinking = True
                elif not current_thinking and is_thinking:
                    print("\n[Answer] ", end="", flush=True)
                    is_thinking = False

                # 提取 Token (参考 grok2api 逻辑)
                token = res_obj.get("token")
                if token and isinstance(token, str):
                    print(token, end="", flush=True)
                    full_text += token
                
                # 检查是否结束
                if res_obj.get("isComplete"):
                    print("\n[+] Response complete.")
                    break
                    
            except json.JSONDecodeError:
                if "<html" in line_str.lower():
                    print(f"\n[!] Cloudflare Verification Detected (HTML Response)")
                    # print(f"DEBUG RAW: {line_str[:200]}...")
                    break
                continue

        if not full_text:
            print("\n[-] No content parsed from stream.")
        else:
            print(f"\n[V] Success! Total characters: {len(full_text)}")

    except Exception as e:
        print(f"\n[-] Request failed: {e}")

if __name__ == "__main__":
    test_p = "你好，请确认你的模型版本，并简单打个招呼。"
    if len(sys.argv) > 1:
        test_p = " ".join(sys.argv[1:])
    test_chat(test_p)