عدّل التعليقات والمتغيرات حسب حاجتك.
Signed-off-by: Mohamed-Abdelhalim2 <mohamed-abdelhalim2@noreply.app.gitpasha.com>
هذا الالتزام موجود في:
242
task_runner.py
Normal file
242
task_runner.py
Normal file
@@ -0,0 +1,242 @@
|
||||
# task_runner.py
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import subprocess
|
||||
import time
|
||||
import shutil
|
||||
import requests
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
# ---------- إعدادات عامة ----------
|
||||
REPO_DIR = Path(os.environ.get("REPO_DIR", ".")) # جذر المستودع
|
||||
TASKS_FILE = REPO_DIR / "tasks.json" # ملف المهام
|
||||
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN") # PAT مطلوب لإنشاء PR أو push لremote محمي
|
||||
GITHUB_API = "https://api.github.com"
|
||||
DEFAULT_BASE_BRANCH = "main" # فرع الأساس عند إنشاء PR
|
||||
|
||||
# ---------- دوال مساعدة ----------
|
||||
def run(cmd, cwd=REPO_DIR, check=True):
|
||||
print(f"> {' '.join(cmd)}")
|
||||
return subprocess.run(cmd, cwd=str(cwd), check=check, text=True, capture_output=True)
|
||||
|
||||
def git_current_branch():
|
||||
res = run(["git", "rev-parse", "--abbrev-ref", "HEAD"], check=True)
|
||||
return res.stdout.strip()
|
||||
|
||||
def ensure_branch(branch):
|
||||
cur = git_current_branch()
|
||||
if cur == branch:
|
||||
return
|
||||
# إذا الفرع موجود محليًا
|
||||
branches = run(["git", "branch", "--list", branch], check=True).stdout
|
||||
if branches.strip():
|
||||
run(["git", "checkout", branch], check=True)
|
||||
else:
|
||||
# إنشاء فرع محلي من الفرع الحالي أو من base
|
||||
run(["git", "checkout", "-b", branch], check=True)
|
||||
|
||||
def commit_and_push(branch, message, push=True, remote="origin"):
|
||||
run(["git", "add", "."], check=True)
|
||||
try:
|
||||
run(["git", "commit", "-m", message], check=True)
|
||||
except subprocess.CalledProcessError:
|
||||
print("لا توجد تغييرات جديدة للالتزام.")
|
||||
return
|
||||
if push:
|
||||
run(["git", "push", "-u", remote, branch], check=True)
|
||||
|
||||
def create_pr(owner, repo, head, base, title, body=""):
|
||||
if not GITHUB_TOKEN:
|
||||
raise RuntimeError("GITHUB_TOKEN غير موجود في متغيرات البيئة.")
|
||||
url = f"{GITHUB_API}/repos/{owner}/{repo}/pulls"
|
||||
headers = {"Authorization": f"token {GITHUB_TOKEN}"}
|
||||
data = {"title": title, "head": head, "base": base, "body": body}
|
||||
r = requests.post(url, json=data, headers=headers)
|
||||
if r.status_code in (200,201):
|
||||
print("PR created:", r.json().get("html_url"))
|
||||
return r.json()
|
||||
else:
|
||||
print("فشل إنشاء PR:", r.status_code, r.text)
|
||||
r.raise_for_status()
|
||||
|
||||
# ---------- تنفيذ مهمة واحدة ----------
|
||||
def execute_task(task):
|
||||
"""
|
||||
نموذج مهمة (task):
|
||||
{
|
||||
"id": "task-1",
|
||||
"type": "create_files" | "create_dirs" | "import_file" | "modify_file" | "merge_branch" | "checkout_branch",
|
||||
"branch": "push-test",
|
||||
"confirm": true,
|
||||
"auto": false,
|
||||
"payload": { ... } # حسب النوع
|
||||
}
|
||||
"""
|
||||
ttype = task.get("type")
|
||||
branch = task.get("branch")
|
||||
confirm = task.get("confirm", True)
|
||||
auto = task.get("auto", False)
|
||||
payload = task.get("payload", {})
|
||||
|
||||
print(f"\n=== تنفيذ المهمة {task.get('id')} نوع: {ttype} ===")
|
||||
|
||||
# تأكيد المستخدم إذا مطلوب
|
||||
if confirm and not auto:
|
||||
ans = input("تنفيذ المهمة الآن؟ (y/N): ").strip().lower()
|
||||
if ans != "y":
|
||||
print("تم إلغاء المهمة بناءً على اختيار المستخدم.")
|
||||
return
|
||||
|
||||
# تأكد من الفرع المطلوب
|
||||
if branch:
|
||||
ensure_branch(branch)
|
||||
|
||||
# أنواع المهام
|
||||
if ttype == "create_dirs":
|
||||
dirs = payload.get("dirs", [])
|
||||
for d in dirs:
|
||||
p = REPO_DIR / d
|
||||
p.mkdir(parents=True, exist_ok=True)
|
||||
print("تم إنشاء/التأكد من وجود:", p)
|
||||
elif ttype == "create_files":
|
||||
files = payload.get("files", [])
|
||||
for f in files:
|
||||
p = REPO_DIR / f["path"]
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
content = f.get("content", "")
|
||||
with open(p, "w", encoding="utf-8") as fh:
|
||||
fh.write(content)
|
||||
print("تم إنشاء/تعديل الملف:", p)
|
||||
elif ttype == "import_file":
|
||||
# import from local path to repo
|
||||
src = Path(payload.get("src"))
|
||||
dest = REPO_DIR / payload.get("dest", src.name)
|
||||
if not src.exists():
|
||||
print("ملف المصدر غير موجود:", src)
|
||||
else:
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
shutil.copy2(src, dest)
|
||||
print(f"تم نسخ {src} إلى {dest}")
|
||||
elif ttype == "modify_file":
|
||||
path = REPO_DIR / payload.get("path")
|
||||
if not path.exists():
|
||||
print("الملف غير موجود:", path)
|
||||
else:
|
||||
# يمكن أن يكون التعديل عبر استبدال كامل أو إضافة سطور
|
||||
mode = payload.get("mode", "replace") # replace | append
|
||||
content = payload.get("content", "")
|
||||
if mode == "append":
|
||||
with open(path, "a", encoding="utf-8") as fh:
|
||||
fh.write(content)
|
||||
else:
|
||||
with open(path, "w", encoding="utf-8") as fh:
|
||||
fh.write(content)
|
||||
print("تم تعديل الملف:", path)
|
||||
elif ttype == "checkout_branch":
|
||||
target = payload.get("target")
|
||||
if not target:
|
||||
print("لم يتم تحديد اسم الفرع الهدف.")
|
||||
else:
|
||||
ensure_branch(target)
|
||||
print("الانتقال إلى الفرع:", target)
|
||||
elif ttype == "merge_branch":
|
||||
source = payload.get("source")
|
||||
target = payload.get("target", git_current_branch())
|
||||
if not source:
|
||||
print("لم يتم تحديد فرع المصدر للدمج.")
|
||||
else:
|
||||
# انتقل إلى target ثم ادمج source
|
||||
ensure_branch(target)
|
||||
run(["git", "merge", "--no-ff", source], check=True)
|
||||
print(f"تم دمج {source} إلى {target}")
|
||||
else:
|
||||
print("نوع المهمة غير مدعوم:", ttype)
|
||||
return
|
||||
|
||||
# بعد التعديلات: التزام ودفع حسب إعداد المهمة
|
||||
commit_msg = payload.get("commit_message", f"Auto: {task.get('id')}")
|
||||
push = payload.get("push", True)
|
||||
if payload.get("create_branch_for_changes"):
|
||||
# إنشاء فرع جديد باسم مخصص
|
||||
new_branch = payload.get("new_branch_name", f"auto/{int(time.time())}")
|
||||
run(["git", "checkout", "-b", new_branch], check=True)
|
||||
commit_and_push(new_branch, commit_msg, push=push)
|
||||
# إنشاء PR إذا مطلوب
|
||||
if payload.get("create_pr"):
|
||||
owner = payload.get("owner")
|
||||
repo = payload.get("repo")
|
||||
base = payload.get("base", DEFAULT_BASE_BRANCH)
|
||||
title = payload.get("pr_title", commit_msg)
|
||||
create_pr(owner, repo, new_branch, base, title, payload.get("pr_body", ""))
|
||||
else:
|
||||
# التزام على الفرع الحالي
|
||||
cur = git_current_branch()
|
||||
commit_and_push(cur, commit_msg, push=push)
|
||||
|
||||
# ---------- قراءة ملف المهام وتنفيذها ----------
|
||||
def process_tasks_once():
|
||||
if not TASKS_FILE.exists():
|
||||
print("لم يتم العثور على ملف المهام:", TASKS_FILE)
|
||||
return
|
||||
with open(TASKS_FILE, "r", encoding="utf-8") as fh:
|
||||
data = json.load(fh)
|
||||
tasks = data.get("tasks", [])
|
||||
for t in tasks:
|
||||
if t.get("status") in ("done","skipped"):
|
||||
continue
|
||||
try:
|
||||
execute_task(t)
|
||||
t["status"] = "done"
|
||||
t["completed_at"] = datetime.utcnow().isoformat() + "Z"
|
||||
except Exception as e:
|
||||
print("خطأ أثناء تنفيذ المهمة:", e)
|
||||
t["status"] = "error"
|
||||
t["error"] = str(e)
|
||||
# حفظ التغييرات في ملف المهام
|
||||
with open(TASKS_FILE, "w", encoding="utf-8") as fh:
|
||||
json.dump(data, fh, indent=2, ensure_ascii=False)
|
||||
print("انتهى معالجة المهام.")
|
||||
|
||||
# ---------- تشغيل كخدمة مراقبة (اختياري) ----------
|
||||
def watch_tasks(poll_interval=3):
|
||||
last_mtime = TASKS_FILE.stat().st_mtime if TASKS_FILE.exists() else 0
|
||||
print("تشغيل وضع المراقبة على", TASKS_FILE)
|
||||
try:
|
||||
while True:
|
||||
if TASKS_FILE.exists():
|
||||
m = TASKS_FILE.stat().st_mtime
|
||||
if m != last_mtime:
|
||||
print("تغيّر ملف المهام — معالجة...")
|
||||
process_tasks_once()
|
||||
last_mtime = m
|
||||
time.sleep(poll_interval)
|
||||
except KeyboardInterrupt:
|
||||
print("تم إيقاف المراقبة.")
|
||||
|
||||
# ---------- CLI بسيط ----------
|
||||
def print_usage():
|
||||
print("Usage:")
|
||||
print(" python task_runner.py run_once # تنفيذ المهام الموجودة في tasks.json ثم الخروج")
|
||||
print(" python task_runner.py watch # مراقبة tasks.json وتنفيذ المهام الجديدة")
|
||||
print(" python task_runner.py exec <taskfile> # تنفيذ ملف مهمة مفرد (json) ثم الخروج")
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print_usage(); sys.exit(1)
|
||||
cmd = sys.argv[1]
|
||||
if cmd == "run_once":
|
||||
process_tasks_once()
|
||||
elif cmd == "watch":
|
||||
watch_tasks()
|
||||
elif cmd == "exec" and len(sys.argv) == 3:
|
||||
# تنفيذ ملف مهمة مفرد
|
||||
tf = Path(sys.argv[2])
|
||||
if not tf.exists():
|
||||
print("ملف المهمة غير موجود:", tf); sys.exit(1)
|
||||
with open(tf, "r", encoding="utf-8") as fh:
|
||||
task = json.load(fh)
|
||||
execute_task(task)
|
||||
else:
|
||||
print_usage()
|
||||
المرجع في مشكلة جديدة
حظر مستخدم