322 lines
12 KiB
Python
322 lines
12 KiB
Python
from datetime import datetime, timedelta
|
||
|
||
from flask import flash, jsonify, redirect, render_template, request, url_for
|
||
|
||
|
||
def computers_manage_icon_label(*, get_conn, clean_text, safe_next_url):
|
||
payload = request.get_json(silent=True) if request.is_json else request.form
|
||
payload = payload or {}
|
||
source_type = clean_text(payload.get("source_type")) or "computer"
|
||
source_id_raw = clean_text(payload.get("source_id") or payload.get("computer_id"))
|
||
icon_label = clean_text(payload.get("icon_label"))[:120]
|
||
next_url = safe_next_url(payload.get("next"), fallback="index")
|
||
|
||
source_settings = {
|
||
"computer": ("computers", "computer_manage_icons", "computer_id"),
|
||
"client_test": ("client_test_devices", "client_manage_icons", "client_id"),
|
||
}
|
||
if source_type not in source_settings or not source_id_raw.isdigit():
|
||
if request.is_json:
|
||
return jsonify(ok=False, error="invalid computer source"), 400
|
||
return redirect(next_url)
|
||
|
||
source_id = int(source_id_raw)
|
||
source_table, labels_table, source_column = source_settings[source_type]
|
||
conn = get_conn()
|
||
cur = conn.cursor()
|
||
cur.execute(f"SELECT 1 FROM {source_table} WHERE id=%s", (source_id,))
|
||
exists = cur.fetchone() is not None
|
||
if not exists:
|
||
conn.close()
|
||
if request.is_json:
|
||
return jsonify(ok=False, error="computer not found"), 404
|
||
return redirect(next_url)
|
||
|
||
if icon_label:
|
||
cur.execute(
|
||
f"""
|
||
INSERT INTO {labels_table} ({source_column}, icon_label)
|
||
VALUES (%s, %s)
|
||
ON CONFLICT ({source_column})
|
||
DO UPDATE SET icon_label = EXCLUDED.icon_label
|
||
""",
|
||
(source_id, icon_label),
|
||
)
|
||
else:
|
||
cur.execute(
|
||
f"DELETE FROM {labels_table} WHERE {source_column}=%s",
|
||
(source_id,),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
if request.is_json:
|
||
return jsonify(ok=True, icon_label=icon_label)
|
||
return redirect(next_url)
|
||
|
||
|
||
def computers_update_cabinet(*, get_conn, log_equipment_movement):
|
||
cid = request.form.get("id", "").strip()
|
||
cabinet_id = request.form.get("cabinet_id", "").strip()
|
||
if not cid.isdigit():
|
||
flash("Некорректный компьютер")
|
||
return redirect(request.referrer or url_for("index"))
|
||
cab_id = int(cabinet_id) if cabinet_id.isdigit() else None
|
||
conn = get_conn()
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT inventory_number, cabinet_id FROM computers WHERE id=%s", (int(cid),))
|
||
row = cur.fetchone()
|
||
inv_num = row[0] if row else None
|
||
old_cab_id = row[1] if row else None
|
||
cur.execute("UPDATE computers SET cabinet_id=%s WHERE id=%s", (cab_id, int(cid)))
|
||
conn.commit()
|
||
conn.close()
|
||
if inv_num and old_cab_id != cab_id:
|
||
log_equipment_movement(inv_num, "computer", old_cab_id, cab_id)
|
||
flash("Расположение обновлено")
|
||
return redirect(request.referrer or url_for("index"))
|
||
|
||
|
||
def computers_list(
|
||
*,
|
||
get_conn,
|
||
computer_types,
|
||
computer_brands_source,
|
||
cpu_brands,
|
||
memory_types,
|
||
os_options,
|
||
):
|
||
conn = get_conn()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"""
|
||
SELECT c.id, c.inventory_number, c.brand, c.model, c.serial_number, c.type,
|
||
c.cpu_brand, c.cpu_model, c.gpu_model, c.memory_size, c.memory_type,
|
||
c.storage_size, c.motherboard, c.os, c.rustdesk_id, c.rustdesk_password,
|
||
c.hostname, c.ip_address, c.mac_address, c.is_online, c.last_seen, c.user_name,
|
||
c.cabinet_id, cab.name AS cabinet_name,
|
||
c.note, c.date_in_operation, c.date_added
|
||
FROM computers c
|
||
LEFT JOIN cabinets cab ON cab.id = c.cabinet_id
|
||
ORDER BY
|
||
cab.building NULLS LAST,
|
||
CASE
|
||
WHEN NULLIF(REGEXP_REPLACE(COALESCE(cab.cabinet, cab.name, ''), '\D', '', 'g'), '') IS NULL THEN 1
|
||
ELSE 0
|
||
END,
|
||
COALESCE(NULLIF(REGEXP_REPLACE(COALESCE(cab.cabinet, cab.name, ''), '\D', '', 'g'), ''), '0')::INT,
|
||
cab.cabinet NULLS LAST,
|
||
cab.name NULLS LAST,
|
||
c.inventory_number,
|
||
c.id
|
||
"""
|
||
)
|
||
raw_rows = cur.fetchall()
|
||
now = datetime.now()
|
||
offline_after = timedelta(seconds=35)
|
||
rows = []
|
||
for row in raw_rows:
|
||
last_seen = row[20]
|
||
is_online = bool(row[19])
|
||
if last_seen and now - last_seen > offline_after:
|
||
is_online = False
|
||
status_label = "Online" if is_online else "Offline"
|
||
rows.append(tuple(row) + (status_label,))
|
||
total_computers = len(rows)
|
||
laptops_count = 0
|
||
pcs_count = 0
|
||
for row in rows:
|
||
ctype = (row[5] or "").strip().lower()
|
||
if ctype == "laptop":
|
||
laptops_count += 1
|
||
elif ctype == "pc":
|
||
pcs_count += 1
|
||
cur.execute("SELECT id, name FROM cabinets ORDER BY name")
|
||
cabinets = cur.fetchall()
|
||
cur.execute(computer_brands_source)
|
||
computer_brands = [row[0] for row in cur.fetchall()]
|
||
conn.close()
|
||
return render_template(
|
||
"computers.html",
|
||
rows=rows,
|
||
cabinets=cabinets,
|
||
computer_types=computer_types,
|
||
computer_brands=computer_brands,
|
||
cpu_brands=cpu_brands,
|
||
memory_types=memory_types,
|
||
os_options=os_options,
|
||
total_computers=total_computers,
|
||
laptops_count=laptops_count,
|
||
pcs_count=pcs_count,
|
||
)
|
||
|
||
|
||
def computers_add(*, get_conn):
|
||
inventory_number = request.form.get("inventory_number", "").strip()
|
||
brand = request.form.get("brand", "").strip()
|
||
model = request.form.get("model", "").strip()
|
||
serial_number = request.form.get("serial_number", "").strip()
|
||
ctype = request.form.get("type", "").strip()
|
||
cpu_brand = request.form.get("cpu_brand", "").strip()
|
||
cpu_model = request.form.get("cpu_model", "").strip()
|
||
gpu_model = request.form.get("gpu_model", "").strip()
|
||
memory_size = request.form.get("memory_size", "").strip()
|
||
memory_type = request.form.get("memory_type", "").strip()
|
||
storage_size = request.form.get("storage_size", "").strip()
|
||
motherboard = request.form.get("motherboard", "").strip()
|
||
os_name = request.form.get("os", "").strip()
|
||
rustdesk_id = request.form.get("rustdesk_id", "").strip()
|
||
rustdesk_password = request.form.get("rustdesk_password", "").strip()
|
||
user_name = request.form.get("user_name", "").strip()
|
||
cabinet_id = request.form.get("cabinet_id", "").strip()
|
||
note = request.form.get("note", "").strip()
|
||
date_in_operation = request.form.get("date_in_operation", "").strip()
|
||
|
||
if not inventory_number:
|
||
flash("Укажите инвентарный номер")
|
||
return redirect(url_for("computers"))
|
||
if ctype not in ("pc", "laptop"):
|
||
flash("Укажите тип (ПК/ноутбук)")
|
||
return redirect(url_for("computers"))
|
||
|
||
cab_id = int(cabinet_id) if cabinet_id.isdigit() else None
|
||
conn = get_conn()
|
||
cur = conn.cursor()
|
||
try:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO computers (
|
||
inventory_number, brand, model, serial_number, type,
|
||
cpu_brand, cpu_model, gpu_model, memory_size, memory_type,
|
||
storage_size, motherboard, os, rustdesk_id, rustdesk_password,
|
||
user_name, cabinet_id, note, date_in_operation
|
||
)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
""",
|
||
(
|
||
inventory_number,
|
||
brand or None,
|
||
model or None,
|
||
serial_number or None,
|
||
ctype,
|
||
cpu_brand or None,
|
||
cpu_model or None,
|
||
gpu_model or None,
|
||
memory_size or None,
|
||
memory_type or None,
|
||
storage_size or None,
|
||
motherboard or None,
|
||
os_name or None,
|
||
rustdesk_id or None,
|
||
rustdesk_password or None,
|
||
user_name or None,
|
||
cab_id,
|
||
note or None,
|
||
date_in_operation or None,
|
||
),
|
||
)
|
||
conn.commit()
|
||
flash("Компьютер добавлен")
|
||
except Exception:
|
||
conn.rollback()
|
||
flash("Не удалось добавить компьютер")
|
||
finally:
|
||
conn.close()
|
||
return redirect(url_for("computers"))
|
||
|
||
|
||
def computers_edit(*, get_conn, log_equipment_movement):
|
||
cid = request.form.get("id", "").strip()
|
||
inventory_number = request.form.get("inventory_number", "").strip()
|
||
brand = request.form.get("brand", "").strip()
|
||
model = request.form.get("model", "").strip()
|
||
serial_number = request.form.get("serial_number", "").strip()
|
||
ctype = request.form.get("type", "").strip()
|
||
cpu_brand = request.form.get("cpu_brand", "").strip()
|
||
cpu_model = request.form.get("cpu_model", "").strip()
|
||
gpu_model = request.form.get("gpu_model", "").strip()
|
||
memory_size = request.form.get("memory_size", "").strip()
|
||
memory_type = request.form.get("memory_type", "").strip()
|
||
storage_size = request.form.get("storage_size", "").strip()
|
||
motherboard = request.form.get("motherboard", "").strip()
|
||
os_name = request.form.get("os", "").strip()
|
||
rustdesk_id = request.form.get("rustdesk_id", "").strip()
|
||
rustdesk_password = request.form.get("rustdesk_password", "").strip()
|
||
user_name = request.form.get("user_name", "").strip()
|
||
cabinet_id = request.form.get("cabinet_id", "").strip()
|
||
note = request.form.get("note", "").strip()
|
||
date_in_operation = request.form.get("date_in_operation", "").strip()
|
||
|
||
if not cid.isdigit():
|
||
flash("Некорректное устройство")
|
||
return redirect(url_for("computers"))
|
||
if not inventory_number:
|
||
flash("Укажите инвентарный номер")
|
||
return redirect(url_for("computers"))
|
||
if ctype not in ("pc", "laptop"):
|
||
flash("Укажите тип устройства (ПК/ноутбук)")
|
||
return redirect(url_for("computers"))
|
||
|
||
cab_id = int(cabinet_id) if cabinet_id.isdigit() else None
|
||
conn = get_conn()
|
||
cur = conn.cursor()
|
||
try:
|
||
cur.execute("SELECT cabinet_id FROM computers WHERE id=%s", (int(cid),))
|
||
row = cur.fetchone()
|
||
old_cab_id = row[0] if row else None
|
||
cur.execute(
|
||
"""
|
||
UPDATE computers
|
||
SET inventory_number=%s, brand=%s, model=%s, serial_number=%s, type=%s,
|
||
cpu_brand=%s, cpu_model=%s, gpu_model=%s, memory_size=%s, memory_type=%s,
|
||
storage_size=%s, motherboard=%s, os=%s, rustdesk_id=%s, rustdesk_password=%s,
|
||
user_name=%s, cabinet_id=%s, note=%s, date_in_operation=%s
|
||
WHERE id=%s
|
||
""",
|
||
(
|
||
inventory_number,
|
||
brand or None,
|
||
model or None,
|
||
serial_number or None,
|
||
ctype,
|
||
cpu_brand or None,
|
||
cpu_model or None,
|
||
gpu_model or None,
|
||
memory_size or None,
|
||
memory_type or None,
|
||
storage_size or None,
|
||
motherboard or None,
|
||
os_name or None,
|
||
rustdesk_id or None,
|
||
rustdesk_password or None,
|
||
user_name or None,
|
||
cab_id,
|
||
note or None,
|
||
date_in_operation or None,
|
||
int(cid),
|
||
),
|
||
)
|
||
conn.commit()
|
||
if old_cab_id != cab_id:
|
||
log_equipment_movement(inventory_number, "computer", old_cab_id, cab_id)
|
||
flash("Компьютер обновлён")
|
||
except Exception:
|
||
conn.rollback()
|
||
flash("Не удалось обновить компьютер")
|
||
finally:
|
||
conn.close()
|
||
return redirect(url_for("computers"))
|
||
|
||
|
||
def computers_delete(*, get_conn):
|
||
cid = request.form.get("id", "").strip()
|
||
if not cid.isdigit():
|
||
flash("Некорректное устройство")
|
||
return redirect(url_for("computers"))
|
||
conn = get_conn()
|
||
cur = conn.cursor()
|
||
cur.execute("DELETE FROM computers WHERE id=%s", (int(cid),))
|
||
conn.commit()
|
||
conn.close()
|
||
flash("Компьютер удалён")
|
||
return redirect(url_for("computers"))
|