""" tasks.py — صفحة التاسكات (مستقلة) لمشروع server-monitor الصفحات: /tasks إدارة التاسكات /tasks/report تقرير لكل شركة (طباعة / PDF / CSV) الـ API: GET /api/tasks?status=all|active|done&employee=&company=&date_from=&date_to= POST /api/tasks PUT /api/tasks/{id} DELETE /api/tasks/{id} GET /api/tasks/employees GET /api/tasks/companies GET /api/tasks/export.csv?company=&date_from=&date_to= """ import csv import io from datetime import datetime from typing import Optional from fastapi import HTTPException from fastapi.responses import HTMLResponse, Response from pydantic import BaseModel VALID_STATUS = ("pending", "new", "in_progress", "solving", "done") STATUS_AR = { "pending": "لم يبدأ", "new": "جديدة", "in_progress": "يتم العمل عليها", "solving": "جاري حل المشكلة", "done": "تم الانتهاء", } class TaskIn(BaseModel): title: str details: str = "" employee: str = "" company: str = "" start_date: str = "" end_date: str = "" status: str = "new" class TaskUpdate(BaseModel): title: Optional[str] = None details: Optional[str] = None employee: Optional[str] = None company: Optional[str] = None start_date: Optional[str] = None end_date: Optional[str] = None status: Optional[str] = None def _check_dates(start: str, end: str) -> None: for label, v in (("start_date", start), ("end_date", end)): if v: try: datetime.strptime(v, "%Y-%m-%d") except ValueError: raise HTTPException(status_code=400, detail=f"{label} غير صحيح (YYYY-MM-DD)") if start and end and end < start: raise HTTPException(status_code=400, detail="تاريخ الانتهاء لازم يكون بعد تاريخ البداية") def register(app, get_db, standalone=False): conn = get_db() conn.execute( """CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, details TEXT DEFAULT '', employee TEXT DEFAULT '', company TEXT DEFAULT '', start_date TEXT DEFAULT '', end_date TEXT DEFAULT '', status TEXT DEFAULT 'new', created_at TEXT NOT NULL, completed_at TEXT DEFAULT '' )""" ) cols = [r[1] for r in conn.execute("PRAGMA table_info(tasks)").fetchall()] if "company" not in cols: # ترقية الجدول القديم conn.execute("ALTER TABLE tasks ADD COLUMN company TEXT DEFAULT ''") conn.commit() conn.close() def _query(status="all", employee="", company="", date_from="", date_to=""): sql = "SELECT * FROM tasks WHERE 1=1" params = [] if status == "active": sql += " AND status != 'done'" elif status == "done": sql += " AND status = 'done'" if employee: sql += " AND employee = ?" params.append(employee) if company: if company == "__none__": sql += " AND company = ''" else: sql += " AND company = ?" params.append(company) day = "COALESCE(NULLIF(start_date, ''), substr(created_at, 1, 10))" if date_from: sql += f" AND {day} >= ?" params.append(date_from) if date_to: sql += f" AND {day} <= ?" params.append(date_to) sql += """ ORDER BY CASE WHEN status = 'done' THEN 1 ELSE 0 END, CASE WHEN status = 'done' THEN '' WHEN end_date = '' THEN '9999' ELSE end_date END ASC, completed_at DESC, id DESC""" conn = get_db() rows = conn.execute(sql, params).fetchall() conn.close() today = datetime.now().strftime("%Y-%m-%d") out = [] for r in rows: d = dict(r) d["overdue"] = bool(d["status"] != "done" and d["end_date"] and d["end_date"] < today) out.append(d) return out @app.get("/tasks", response_class=HTMLResponse) async def tasks_page(): html = TASKS_HTML if standalone: # موقع تاسكات لوحده: شيل روابط لوحة المراقبة وصفحة الحالة html = html.replace( '📊 تقرير الشركات  |  لوحة المراقبة  |  صفحة الحالة', '📊 تقرير الشركات', ) return HTMLResponse(html) @app.get("/tasks/report", response_class=HTMLResponse) async def tasks_report_page(): html = REPORT_HTML if standalone: html = html.replace( '← التاسكات  |  لوحة المراقبة', '← التاسكات', ) return HTMLResponse(html) @app.get("/api/tasks/employees") def tasks_employees(): conn = get_db() rows = conn.execute( "SELECT DISTINCT employee FROM tasks WHERE employee != '' ORDER BY employee COLLATE NOCASE" ).fetchall() conn.close() return [r["employee"] for r in rows] @app.get("/api/tasks/companies") def tasks_companies(): conn = get_db() rows = conn.execute( "SELECT DISTINCT company FROM tasks WHERE company != '' ORDER BY company COLLATE NOCASE" ).fetchall() conn.close() return [r["company"] for r in rows] @app.get("/api/tasks/export.csv") def tasks_export(company: str = "", date_from: str = "", date_to: str = ""): rows = _query("all", "", company, date_from, date_to) buf = io.StringIO() w = csv.writer(buf) w.writerow(["الشركة", "العنوان", "التفاصيل", "الموظف", "الحالة", "البداية", "الانتهاء", "تاريخ الإنجاز", "متأخرة"]) for r in sorted(rows, key=lambda x: (x["company"] or "~", x["id"])): w.writerow([ r["company"], r["title"], r["details"], r["employee"], STATUS_AR.get(r["status"], r["status"]), r["start_date"], r["end_date"], r["completed_at"], "نعم" if r["overdue"] else "", ]) data = ("\ufeff" + buf.getvalue()).encode("utf-8") # BOM علشان Excel يقرا العربي صح return Response( content=data, media_type="text/csv; charset=utf-8", headers={"Content-Disposition": "attachment; filename=tasks-report.csv"}, ) @app.get("/api/tasks") def tasks_list(status: str = "all", employee: str = "", company: str = "", date_from: str = "", date_to: str = ""): return _query(status, employee, company, date_from, date_to) @app.post("/api/tasks") def tasks_create(p: TaskIn): title = p.title.strip() if not title: raise HTTPException(status_code=400, detail="العنوان مطلوب") if p.status not in VALID_STATUS: raise HTTPException(status_code=400, detail="حالة غير صحيحة") _check_dates(p.start_date, p.end_date) now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") conn = get_db() cur = conn.execute( """INSERT INTO tasks (title, details, employee, company, start_date, end_date, status, created_at, completed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (title, p.details.strip(), p.employee.strip(), p.company.strip(), p.start_date, p.end_date, p.status, now, now if p.status == "done" else ""), ) conn.commit() task_id = cur.lastrowid conn.close() return {"ok": True, "id": task_id} @app.put("/api/tasks/{task_id}") def tasks_update(task_id: int, p: TaskUpdate): conn = get_db() row = conn.execute("SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() if not row: conn.close() raise HTTPException(status_code=404, detail="التاسك غير موجود") cur = dict(row) for k, v in p.model_dump(exclude_none=True).items(): cur[k] = v.strip() if isinstance(v, str) else v try: if not cur["title"]: raise HTTPException(status_code=400, detail="العنوان مطلوب") if cur["status"] not in VALID_STATUS: raise HTTPException(status_code=400, detail="حالة غير صحيحة") _check_dates(cur["start_date"], cur["end_date"]) except HTTPException: conn.close() raise if cur["status"] == "done": if row["status"] != "done": cur["completed_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") else: cur["completed_at"] = "" conn.execute( """UPDATE tasks SET title=?, details=?, employee=?, company=?, start_date=?, end_date=?, status=?, completed_at=? WHERE id=?""", (cur["title"], cur["details"], cur["employee"], cur["company"], cur["start_date"], cur["end_date"], cur["status"], cur["completed_at"], task_id), ) conn.commit() conn.close() return {"ok": True} @app.delete("/api/tasks/{task_id}") def tasks_delete(task_id: int): conn = get_db() conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,)) conn.commit() conn.close() return {"ok": True} TASKS_HTML = """ التاسكات | Tasks

📋 التاسكات

📊 تقرير الشركات  |  لوحة المراقبة  |  صفحة الحالة
0شغالة
0متأخرة
0منتهية
0الإجمالي

تاسك جديد

""" REPORT_HTML = """ تقرير التاسكات حسب الشركة

📊 تقرير التاسكات حسب الشركة

← التاسكات  |  لوحة المراقبة
⬇ تصدير Excel (CSV)

تقرير التاسكات

"""