tags only. No , , tags."
}}
Focus keyword to use naturally in title, first paragraph, and headings: {keyword}
"""
url = "https://api.groq.com/openai/v1/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": "You are a helpful assistant that strictly outputs valid JSON only."},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_completion_tokens": 3000,
"response_format": {"type": "json_object"}
}
log(f"Calling Groq API for keyword: '{keyword}' in language: {lang}")
resp = requests.post(url, headers=headers, json=payload, timeout=60)
resp.raise_for_status()
raw = resp.json()
text = raw["choices"][0]["message"]["content"].strip()
article = json.loads(text)
# === إضافة روابط التطبيقات الثابتة في آخر المقال ===
if lang == "ar":
cta_html = f"""
📱 احصل على تطبيق {site} الآن!
عزز رحلتك في التحضير لاختبار ALCPT وإتقان اللغة الإنجليزية من خلال تطبيقنا الرسمي:
"""
else:
cta_html = f"""
📱 Get the {site} App Now!
Enhance your ALCPT preparation and English proficiency journey with our official app:
"""
article["content"] += cta_html
# ====================================================
log(f"Article generated ({lang}): '{article['title']}'")
return article
# ═══════════════════════════════════════════════════════════
# 4. جلب صورة من Unsplash (مجاني)
# ═══════════════════════════════════════════════════════════
def fetch_unsplash_image(access_key: str, keyword: str) -> Optional[bytes]:
"""
يجلب صورة مجانية من Unsplash بناءً على الـ keyword.
يُرجع: bytes (بيانات الصورة) أو None لو فشل.
"""
search_query = keyword.replace("/", " ").replace("quiz", "").strip()
# استخدام كلمات بحث أعم لنتائج أفضل
if "ALCPT" in search_query or "ALC" in search_query:
search_query = "English language learning military education"
url = "https://api.unsplash.com/photos/random"
params = {
"query": search_query,
"orientation": "landscape",
"count": 1
}
headers = {"Authorization": f"Client-ID {access_key}"}
log(f"Fetching Unsplash image for: '{search_query}'")
try:
resp = requests.get(url, params=params, headers=headers, timeout=30)
resp.raise_for_status()
photos = resp.json()
photo = photos[0] if isinstance(photos, list) else photos
img_url = photo["urls"]["regular"]
img_resp = requests.get(img_url, timeout=30)
img_resp.raise_for_status()
log("Unsplash image downloaded successfully.")
return img_resp.content
except Exception as e:
log(f"Unsplash fetch failed: {e}", "WARNING")
return None
# ═══════════════════════════════════════════════════════════
# 5. رفع الصورة لـ WordPress Media
# ═══════════════════════════════════════════════════════════
def upload_wp_media(cfg: dict, image_bytes: bytes, filename: str) -> Optional[int]:
"""
يرفع الصورة لـ WordPress Media Library.
يُرجع: media_id أو None لو فشل.
"""
wp_url = cfg["wordpress"]["url"].rstrip("/")
user = cfg["wordpress"]["username"]
app_pwd = cfg["wordpress"]["app_password"]
credentials = base64.b64encode(f"{user}:{app_pwd}".encode()).decode()
headers = {
"Authorization": f"Basic {credentials}",
"Content-Disposition": f'attachment; filename="{filename}"',
"Content-Type": "image/jpeg"
}
endpoint = f"{wp_url}/wp-json/wp/v2/media"
log(f"Uploading media to WordPress: {filename}")
try:
resp = requests.post(endpoint, headers=headers, data=image_bytes, timeout=60)
resp.raise_for_status()
media_id = resp.json()["id"]
log(f"Media uploaded with ID: {media_id}")
return media_id
except Exception as e:
log(f"Media upload failed: {e}", "WARNING")
return None
# ═══════════════════════════════════════════════════════════
# 6. نشر البوست على WordPress
# ═══════════════════════════════════════════════════════════
def publish_wp_post(cfg: dict, article: dict, media_id: Optional[int], keyword: str) -> dict:
"""
ينشر البوست على WordPress عبر REST API.
يُرجع: بيانات البوست المنشور.
"""
wp_url = cfg["wordpress"]["url"].rstrip("/")
user = cfg["wordpress"]["username"]
app_pwd = cfg["wordpress"]["app_password"]
status = cfg["wordpress"].get("post_status", "publish")
category = cfg["wordpress"].get("default_category", 1)
credentials = base64.b64encode(f"{user}:{app_pwd}".encode()).decode()
headers = {
"Authorization": f"Basic {credentials}",
"Content-Type": "application/json"
}
payload = {
"title": article["title"],
"content": article["content"],
"excerpt": article.get("excerpt", ""),
"status": status,
"categories": [category],
"tags": [],
"meta": {}
}
# إضافة الصورة البارزة لو موجودة
if media_id:
payload["featured_media"] = media_id
endpoint = f"{wp_url}/wp-json/wp/v2/posts"
log(f"Publishing post to WordPress: '{article['title']}'")
resp = requests.post(endpoint, headers=headers, json=payload, timeout=60)
resp.raise_for_status()
post_data = resp.json()
post_url = post_data.get("link", "")
post_id = post_data.get("id", 0)
log(f"✅ Post published! ID: {post_id} | URL: {post_url}")
return post_data
# ═══════════════════════════════════════════════════════════
# 7. تحديث ملف التتبع
# ═══════════════════════════════════════════════════════════
def update_progress(progress: dict, keyword: str, index: int, post_id: int, post_url: str):
"""يحدّث progress.json بعد كل نشر ناجح."""
progress["last_index"] = index
progress["last_run"] = datetime.now().isoformat()
progress["stats"]["total_published"] = progress["stats"].get("total_published", 0) + 1
progress["published_posts"].append({
"index": index,
"keyword": keyword,
"post_id": post_id,
"post_url": post_url,
"date": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
})
return progress
def send_admin_notification(cfg: dict, message: str):
"""إرسال إشعارات عبر اليوزربوت (باسم @haarods)."""
# البحث عن إعدادات اليوزربوت في ملف إعدادات السوشيال (لأنها ليست في config.json الخاص بالووردبريس)
social_config_path = os.path.abspath(os.path.join(SCRIPT_DIR, "..", "social_auto_poster", "social_config.json"))
if not os.path.exists(social_config_path): return
with open(social_config_path, "r", encoding="utf-8") as f:
social_cfg = json.load(f)
tg_u = social_cfg.get("telegram_userbot", {})
if not tg_u.get("enabled"): return
api_id = tg_u.get("api_id")
api_hash = tg_u.get("api_hash")
targets = ["@alcpt_practice", "@haarods"]
if not api_id or not api_hash: return
import asyncio
try:
from telethon import TelegramClient
except ImportError:
return
async def _send():
# نستخدم نفس مسار الجلسة الموجود في مجلد السوشيال ميديا لتوحيد الدخول
session_path = os.path.abspath(os.path.join(SCRIPT_DIR, "..", "social_auto_poster", "lexicore_userbot"))
client = TelegramClient(session_path, api_id, api_hash)
await client.connect()
if not await client.is_user_authorized():
return
for target in targets:
try:
await client.send_message(target, message, parse_mode='html')
except:
pass
await client.disconnect()
try:
loop = asyncio.get_event_loop()
if loop.is_running():
asyncio.ensure_future(_send())
else:
loop.run_until_complete(_send())
except:
pass
# ═══════════════════════════════════════════════════════════
# 8. التحقق من الإعدادات
# ═══════════════════════════════════════════════════════════
def validate_config(cfg: dict) -> bool:
"""يتحقق أن جميع المطلوبات موجودة في config.json."""
errors = []
if "YOUR_GROQ_API_KEY_HERE" in cfg.get("groq", {}).get("api_key", "") or not cfg.get("groq", {}).get("api_key"):
errors.append("❌ Groq API Key غير مضبوطة في config.json")
if "YOUR_WP_USERNAME" in cfg["wordpress"]["username"]:
errors.append("❌ WordPress Username غير مضبوط في config.json")
if "YOUR_WP_APPLICATION_PASSWORD" in cfg["wordpress"]["app_password"]:
errors.append("❌ WordPress Application Password غير مضبوط في config.json")
if errors:
for e in errors:
log(e, "ERROR")
return False
return True
# ═══════════════════════════════════════════════════════════
# 9. الدالة الرئيسية
# ═══════════════════════════════════════════════════════════
def main():
parser = argparse.ArgumentParser(description="WordPress Auto-Post (Free)")
parser.add_argument("--test", action="store_true", help="وضع الاختبار: ينشر كـ draft")
parser.add_argument("--keyword", type=str, help="تحديد keyword يدوياً")
parser.add_argument("--lang", type=str, choices=["en", "ar"], help="لغة المقال (en أو ar)")
parser.add_argument("--dry-run", action="store_true", help="توليد المقال فقط بدون نشر")
parser.add_argument("--reset", action="store_true", help="إعادة تعيين progress.json من البداية")
args = parser.parse_args()
log("=" * 55)
log("WordPress Auto-Post — learn-lexicore.bytronex.com")
log("=" * 55)
# ── تحميل الإعدادات ──
if not os.path.exists(CONFIG_FILE):
log("config.json غير موجود! نفّذ: cp config.json.example config.json", "ERROR")
sys.exit(1)
cfg = load_json(CONFIG_FILE)
progress = load_json(PROGRESS_FILE) if os.path.exists(PROGRESS_FILE) else {
"last_index": -1, "published_posts": [], "last_run": None,
"stats": {"total_published": 0, "total_failed": 0}
}
keywords = load_keywords()
log(f"Keywords loaded: {len(keywords)} total")
log(f"Published so far: {progress['stats'].get('total_published', 0)}")
# ── إعادة التعيين ──
if args.reset:
progress = {"last_index": -1, "published_posts": [], "last_run": None,
"stats": {"total_published": 0, "total_failed": 0}}
save_json(PROGRESS_FILE, progress)
log("✅ Progress reset to beginning.")
return
# ── التحقق من الإعدادات ──
# في dry-run نتحقق من Groq فقط (لا نحتاج WordPress)
if args.dry_run:
if "YOUR_GROQ_API_KEY_HERE" in cfg.get("groq", {}).get("api_key", ""):
log("❌ Groq API Key غير مضبوطة في config.json", "ERROR")
sys.exit(1)
elif not validate_config(cfg):
log("عدّل config.json أولاً قبل التشغيل.", "ERROR")
sys.exit(1)
# ── وضع الاختبار: يجبر draft ──
if args.test:
cfg["wordpress"]["post_status"] = "draft"
log("[TEST MODE] Post will be saved as DRAFT.")
# ── لغة المقال ──
lang = args.lang if args.lang else cfg["settings"].get("article_language", "en")
# ── اختيار الـ Keyword ──
if args.keyword:
keyword = args.keyword
index = progress.get("last_index", -1) # لا يحرّك الـ index
log(f"Manual keyword: '{keyword}'")
else:
keyword, index = pick_next_keyword(keywords, progress)
log(f"Auto keyword [{index+1}/{len(keywords)}]: '{keyword}'")
# ── توليد المقال ──
try:
article = generate_article(cfg, keyword, lang)
except Exception as e:
log(f"Failed to generate article: {e}", "ERROR")
progress["stats"]["total_failed"] = progress["stats"].get("total_failed", 0) + 1
save_json(PROGRESS_FILE, progress)
sys.exit(1)
if args.dry_run:
log("[DRY RUN] Article generated. No post published.")
print("\n--- TITLE ---")
print(article["title"])
print("\n--- EXCERPT ---")
print(article.get("excerpt", ""))
print("\n--- CONTENT (first 500 chars) ---")
print(article["content"][:500] + "...")
return
# ── جلب الصورة (مجانية) ──
media_id = None
unsplash_key = cfg.get("unsplash", {}).get("access_key", "")
if unsplash_key and "YOUR_UNSPLASH_ACCESS_KEY_HERE" not in unsplash_key:
img_bytes = fetch_unsplash_image(unsplash_key, keyword)
if img_bytes:
safe_name = keyword.replace(" ", "_").replace("/", "_")[:40]
filename = f"{safe_name}_{datetime.now().strftime('%Y%m%d')}.jpg"
media_id = upload_wp_media(cfg, img_bytes, filename)
else:
log("Unsplash key not set — skipping featured image.", "WARNING")
# ── نشر على WordPress ──
try:
post_data = publish_wp_post(cfg, article, media_id, keyword)
post_id = post_data.get("id", 0)
post_url = post_data.get("link", "")
# ── تحديث التتبع ──
progress = update_progress(progress, keyword, index, post_id, post_url)
save_json(PROGRESS_FILE, progress)
remaining = len(keywords) - (index + 1)
log(f"🎉 Done! Keywords remaining: {remaining}")
log(f" Post URL: {post_url}")
# الإشعار
lang_str = "الإنجليزية" if lang == "en" else "العربية"
notif_msg = f"📝 تم نشر مقال جديد على ووردبريس ({lang_str})!\n\n" \
f"📌 العنوان: {article['title']}\n" \
f"🔑 الكلمة: {keyword}\n" \
f"🔗 الرابط: {post_url}"
send_admin_notification(cfg, notif_msg)
except Exception as e:
log(f"Failed to publish post: {e}", "ERROR")
progress["stats"]["total_failed"] = progress["stats"].get("total_failed", 0) + 1
save_json(PROGRESS_FILE, progress)
sys.exit(1)
if __name__ == "__main__":
main()