Commit 2fe68ad3 authored by Vũ Hoàng Anh's avatar Vũ Hoàng Anh

feat: add n8n ultra description API + product web desc generator

parent 37a3151d
This diff is collapsed.
This diff is collapsed.
"""
Test script: Kiểm tra fetch ảnh từ URL Canifa → base64
Mục đích: Chứng minh API tự lấy ảnh từ URL, n8n KHÔNG cần gửi base64.
Chạy: python -m backend.api.product_desc.test.test_fetch_image
hoặc: python backend/api/product_desc/test/test_fetch_image.py
"""
import httpx
import base64
import sys
# ─── Sample data từ Google Sheets ────────────────────────────────────────────
SAMPLE_IMAGE_URL = "http://mdp.canifa.com/images-web/cache/c8/00/c80079b26cd5437c1473d98930fc1a07.jpg"
def test_fetch_image():
print("=" * 60)
print("🧪 TEST: Fetch ảnh từ URL Canifa → Base64")
print("=" * 60)
print(f"\n📎 URL: {SAMPLE_IMAGE_URL}")
try:
# Fetch ảnh — giống hệt logic trong product_generate_desc_web.py
r = httpx.get(
SAMPLE_IMAGE_URL,
headers={"User-Agent": "Mozilla/5.0"},
timeout=15,
follow_redirects=True,
)
r.raise_for_status()
# Thông tin response
content_type = r.headers.get("content-type", "unknown")
size_bytes = len(r.content)
size_kb = round(size_bytes / 1024, 1)
print(f"\n✅ Fetch thành công!")
print(f" Status code : {r.status_code}")
print(f" Content-Type: {content_type}")
print(f" Size : {size_bytes:,} bytes ({size_kb} KB)")
# Convert base64
b64 = base64.b64encode(r.content).decode()
b64_len = len(b64)
b64_preview = b64[:80] + "..."
print(f"\n🔄 Base64 encode:")
print(f" Length : {b64_len:,} chars")
print(f" Preview : {b64_preview}")
# Validate — check JPEG magic bytes (FFD8FF)
is_jpeg = r.content[:3] == b'\xff\xd8\xff'
is_png = r.content[:4] == b'\x89PNG'
fmt = "JPEG" if is_jpeg else ("PNG" if is_png else "Unknown")
print(f"\n🖼️ Format: {fmt}")
# Simulate Groq API payload
data_url = f"data:{content_type};base64,{b64[:20]}..."
print(f"\n📦 Data URL cho AI Vision:")
print(f" {data_url}")
print(f"\n{'=' * 60}")
print(f"✅ KẾT LUẬN: n8n KHÔNG cần gửi base64!")
print(f" → Chỉ cần gửi 'image_url' (Link ảnh từ Sheets)")
print(f" → API tự fetch + convert base64 server-side")
print(f" → Rồi gửi cho AI Vision model (Llama 4 Scout)")
print(f"{'=' * 60}")
return True
except httpx.HTTPStatusError as e:
print(f"\n❌ HTTP Error: {e.response.status_code}")
print(f" URL có thể bị chặn hoặc không tồn tại")
return False
except httpx.ConnectError as e:
print(f"\n❌ Connection Error: {e}")
print(f" Không kết nối được tới mdp.canifa.com")
return False
except Exception as e:
print(f"\n❌ Error: {type(e).__name__}: {e}")
return False
if __name__ == "__main__":
ok = test_fetch_image()
sys.exit(0 if ok else 1)
"""
Test script: Goi API /api/product-desc/n8n-generate
Simulate n8n HTTP Request node gui data tu Google Sheets.
Chay: py backend/api/product_desc/test/test_n8n_desc.py
"""
import httpx
import json
import sys
import time
# Fix encoding for Windows
if sys.platform == "win32":
sys.stdout.reconfigure(encoding="utf-8")
API_URL = "http://127.0.0.1:5000/api/product-desc/n8n-generate"
# ── Sample data giong het n8n gui tu Google Sheets ──
SAMPLE_PAYLOAD = {
"ma_san_pham": "UNKNOWN",
"ten_san_pham": "Ao ba lo",
"image_url": "https://mdp.canifa.com/images-web/cache/e5/9b/e59b8c724757b142b3eb0d6ebf8d8f7f.jpg",
"thanh_phan": "",
"dac_tinh_chat_lieu": "",
"mo_ta_chung": "",
"tinh_nang": "",
"kieu_dang": "",
"hoan_canh": "",
"featuring": "",
"chung_chi": "",
"huong_dan_su_dung": "",
}
def main():
print("=" * 60)
print("TEST: n8n Ultra Description API")
print("=" * 60)
# Step 1: Check server is running
print("\n[1/3] Checking server...")
try:
r = httpx.get("http://127.0.0.1:5000/health", timeout=5)
if r.status_code == 200:
print(" Server OK")
else:
print(f" Server returned {r.status_code}")
sys.exit(1)
except httpx.ConnectError:
print(" Server NOT running at http://127.0.0.1:5000")
print(" -> Start server first: py backend/server.py")
sys.exit(1)
# Step 2: Send request
print(f"\n[2/3] POST {API_URL}")
print(f" Product: {SAMPLE_PAYLOAD['ten_san_pham']}")
print(f" Image URL: {SAMPLE_PAYLOAD['image_url'][:60]}...")
t0 = time.time()
try:
r = httpx.post(API_URL, json=SAMPLE_PAYLOAD, timeout=120)
elapsed = round(time.time() - t0, 2)
except Exception as e:
print(f" ERROR: {e}")
sys.exit(1)
# Step 3: Parse response
print(f"\n[3/3] Response (took {elapsed}s)")
print(f" Status: {r.status_code}")
try:
data = r.json()
except Exception:
print(f" Raw: {r.text[:500]}")
sys.exit(1)
if data.get("status") == "success":
print(f" Product: {data.get('ten_san_pham')}")
print(f" Server elapsed: {data.get('elapsed_s')}s")
ai_desc = data.get("ai_description", "")
print(f"\n{'=' * 60}")
print("AI DESCRIPTION:")
print("=" * 60)
print(ai_desc)
print(f"\n{'=' * 60}")
print(f"Length: {len(ai_desc)} chars")
# Vision debug
vd = data.get("vision_data", {})
if vd:
print(f"\nVision data: {json.dumps(vd, ensure_ascii=False, indent=2)}")
print(f"\nKET QUA: OK")
else:
print(f" ERROR: {data.get('message')}")
if data.get("raw"):
print(f" Raw: {data['raw'][:300]}")
sys.exit(1)
if __name__ == "__main__":
main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment