Mohamed-Abdelhalim2
02a3a78251
Signed-off-by: Mohamed-Abdelhalim2 <mohamed-abdelhalim2@noreply.app.gitpasha.com>
243 أسطر
9.5 KiB
Python
243 أسطر
9.5 KiB
Python
# 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()
|