File size: 3,449 Bytes
6b78c0d 6606435 d6dd847 6606435 7c86a1d 6b78c0d d6dd847 6606435 d6dd847 6b78c0d d6dd847 6606435 d6dd847 6b78c0d 6606435 d6dd847 6b78c0d d6dd847 6606435 6b78c0d 6606435 d6dd847 6b78c0d 6606435 d6dd847 6606435 6b78c0d d6dd847 6b78c0d 6606435 6b78c0d 6606435 d6dd847 6b78c0d 6606435 d6dd847 6b78c0d d6dd847 6b78c0d d6dd847 6606435 7c86a1d d6dd847 7c86a1d d6dd847 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
import os
import tempfile
import requests
import gradio as gr
from fastapi import FastAPI, Form
from instagrapi import Client
from PIL import Image
import uvicorn
# FastAPI app (for API)
app = FastAPI()
# Instagram client
cl = Client()
# Instagram credentials from Hugging Face Secrets
IG_USERNAME = os.getenv("USERNAME")
IG_PASSWORD = os.getenv("PASSWORD")
# Store logs in memory
logs = []
def log(message: str):
print(message)
logs.append(message)
if len(logs) > 50: # keep last 50 logs
logs.pop(0)
def login():
try:
cl.login(IG_USERNAME, IG_PASSWORD)
log("✅ Logged into Instagram as " + IG_USERNAME)
except Exception as e:
log(f"⚠️ Login failed: {e}")
raise
def fix_google_drive_url(url: str) -> str:
if "drive.google.com" in url:
if "/d/" in url:
file_id = url.split("/d/")[1].split("/")[0]
return f"https://drive.google.com/uc?export=download&id={file_id}"
return url
def process_upload(photo_url: str, caption: str):
try:
if not cl.user_id:
login()
photo_url = fix_google_drive_url(photo_url)
log(f"📥 Downloading from {photo_url}")
response = requests.get(photo_url, stream=True)
if response.status_code != 200:
msg = f"Invalid image URL: {response.status_code}"
log("❌ " + msg)
return {"status": "error", "message": msg}
tmp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".jpg")
tmp_file.write(response.content)
tmp_file.close()
try:
Image.open(tmp_file.name).verify()
except Exception:
msg = "Downloaded file is not a valid image."
log("❌ " + msg)
return {"status": "error", "message": msg}
cl.photo_upload(tmp_file.name, caption)
log("✅ Photo uploaded successfully with caption: " + caption)
return {"status": "success", "message": "Photo uploaded successfully!"}
except Exception as e:
log("⚠️ Exception: " + str(e))
return {"status": "error", "message": str(e)}
# ----------------------------
# FastAPI Endpoint (for Sheets)
# ----------------------------
@app.post("/upload")
def upload_post(photo_url: str = Form(...), caption: str = Form(...)):
return process_upload(photo_url, caption)
@app.get("/")
def root():
return {"status": "ok", "message": "Instagram uploader is running 🚀"}
# ----------------------------
# Gradio UI (for testing/debug)
# ----------------------------
def gradio_upload(photo_url, caption):
result = process_upload(photo_url, caption)
return result["status"], result["message"], "\n".join(logs)
with gr.Blocks() as demo:
gr.Markdown("## 📸 Instagram Auto Uploader")
with gr.Row():
photo_url_in = gr.Textbox(label="Photo URL", placeholder="Paste image link (Google Drive supported)")
caption_in = gr.Textbox(label="Caption", placeholder="Enter your caption")
upload_btn = gr.Button("🚀 Upload to Instagram")
status_out = gr.Textbox(label="Status")
message_out = gr.Textbox(label="Message")
logs_out = gr.Textbox(label="Logs (last 50)", lines=15)
upload_btn.click(fn=gradio_upload, inputs=[photo_url_in, caption_in], outputs=[status_out, message_out, logs_out])
# 🔹 Run Gradio + FastAPI
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", server_port=7860)
|